update api calls for sqlalchemy 2.0

Signed-off-by: Erik Erlandson <[email protected]>

osc_ingest_trino/trino_utils.py

@@ -47,7 +47,8 @@ def _do_sql(sql, engine, verbose=False):
         sql = text(str(sql))
     if verbose:
         print(sql)
-    qres = engine.execute(sql)
+    with engine.begin() as cxn:
+        qres = cxn.execute(sql)
     res = None
     if qres.returns_rows:
         res = qres.fetchall()

osc_ingest_trino/unmanaged/unmanaged_hive_ingest.py

@@ -2,6 +2,7 @@
 import uuid
 
 import pandas as pd
+from sqlalchemy import text
 
 from osc_ingest_trino import create_table_schema_pairs, upload_directory_to_s3
 
@@ -29,8 +30,9 @@ def _prefix(pfx, schema, table):
 
 
 def drop_unmanaged_table(catalog, schema, table, engine, bucket, prefix=_default_prefix, verbose=False):
-    sql = f"drop table if exists {catalog}.{schema}.{table}"
-    qres = engine.execute(sql)
+    sql = text(f"drop table if exists {catalog}.{schema}.{table}")
+    with engine.begin() as cxn:
+        qres = cxn.execute(sql)
     dres = bucket.objects.filter(Prefix=f"{_prefix(prefix, schema, table)}/").delete()
     if verbose:
         print(dres)