DDL execution method for SQLA2

Author: RudolfCardinal

cardinal_pythonlib/sqlalchemy/dump.py

@@ -265,8 +265,9 @@ def dump_orm_object_as_insert_sql(
         core_pkcol = table.columns.get(orm_pkcol.name)
         pkval = getattr(obj, orm_pkcol.name)
         query = query.where(core_pkcol == pkval)
-    cursor = engine.execute(query)
-    row = cursor.fetchone()  # should only be one...
+    with engine.begin() as connection:
+        cursor = connection.execute(query)
+        row = cursor.fetchone()  # should only be one...
     row_dict = dict(row)
     statement = table.insert(values=row_dict)
     insert_str = get_literal_query(statement, bind=engine)

cardinal_pythonlib/sqlalchemy/engine_func.py

@@ -104,8 +104,9 @@ def get_sqlserver_product_version(engine: "Engine") -> Tuple[int, ...]:
         "instances."
     )
     sql = "SELECT CAST(SERVERPROPERTY('ProductVersion') AS VARCHAR)"
-    rp = engine.execute(sql)  # type: Result
-    row = rp.fetchone()
+    with engine.begin() as connection:
+        rp = connection.execute(sql)  # type: Result
+        row = rp.fetchone()
     dotted_version = row[0]  # type: str  # e.g. '12.0.5203.0'
     return tuple(int(x) for x in dotted_version.split("."))
 

cardinal_pythonlib/sqlalchemy/orm_schema.py

@@ -29,11 +29,13 @@
 import logging
 from typing import Type, TYPE_CHECKING
 
-from cardinal_pythonlib.sqlalchemy.session import get_safe_url_from_engine
 from sqlalchemy.engine.base import Engine
 from sqlalchemy.orm import DeclarativeMeta
 from sqlalchemy.schema import CreateTable
 
+from cardinal_pythonlib.sqlalchemy.schema import execute_ddl
+from cardinal_pythonlib.sqlalchemy.session import get_safe_url_from_engine
+
 if TYPE_CHECKING:
     from sqlalchemy.sql.schema import Table
 
@@ -73,5 +75,4 @@ def create_table_from_orm_class(
     creator = CreateTable(
         table, include_foreign_key_constraints=include_foreign_key_constraints
     )
-    with engine.begin() as conn:  # though DML/DDL doesn't really need a COMMIT
-        conn.execute(creator)
+    execute_ddl(engine, ddl=creator)

cardinal_pythonlib/sqlalchemy/schema.py

@@ -59,6 +59,7 @@
     Table,
 )
 from sqlalchemy.sql import sqltypes, text
+from sqlalchemy.sql.ddl import DDLElement
 from sqlalchemy.sql.sqltypes import BigInteger, TypeEngine
 from sqlalchemy.sql.visitors import Visitable
 
@@ -337,6 +338,33 @@ def get_effective_int_pk_col(table_: Table) -> Optional[str]:
     )
 
 
+# =============================================================================
+# Execute DDL
+# =============================================================================
+
+
+def execute_ddl(
+    engine: Engine, sql: str = None, ddl: DDLElement = None
+) -> None:
+    """
+    Execute DDL, either from a plain SQL string, or from an SQLAlchemy DDL
+    element.
+
+    Previously we would use DDL(sql, bind=engine).execute(), but this has gone
+    in SQLAlchemy 2.0.
+
+    If you want dialect-conditional execution, create the DDL object with e.g.
+    ddl = DDL(sql).execute_if(dialect=SqlaDialectName.SQLSERVER), and pass that
+    DDL object to this function.
+    """
+    assert bool(sql) ^ bool(ddl)  # one or the other.
+    if sql:
+        ddl = DDL(sql)
+    with engine.connect() as connection:
+        # DDL doesn't need a COMMIT.
+        connection.execute(ddl)
+
+
 # =============================================================================
 # Indexes
 # =============================================================================
@@ -565,8 +593,7 @@ def quote(identifier: str) -> str:
                     colnames=", ".join(quote(c) for c in colnames),
                 )
             )
-            # DDL(sql, bind=engine).execute_if(dialect=SqlaDialectName.MYSQL)
-            DDL(sql, bind=engine).execute()
+            execute_ddl(engine, sql=sql)
 
         elif is_mssql:  # Microsoft SQL Server
             # https://msdn.microsoft.com/library/ms187317(SQL.130).aspx
@@ -628,7 +655,7 @@ def quote(identifier: str) -> str:
                 )
                 # Executing serial COMMITs or a ROLLBACK won't help here if
                 # this transaction is due to Python DBAPI default behaviour.
-            DDL(sql, bind=engine).execute()
+            execute_ddl(engine, sql=sql)
 
             # The reversal procedure is DROP FULLTEXT INDEX ON tablename;
 

cardinal_pythonlib/sqlalchemy/sqlserver.py

@@ -36,6 +36,7 @@
     quote_identifier,
     SqlaDialectName,
 )
+from cardinal_pythonlib.sqlalchemy.schema import execute_ddl
 from cardinal_pythonlib.sqlalchemy.session import get_engine_from_session
 
 
@@ -48,7 +49,8 @@ def _exec_ddl_if_sqlserver(engine: Engine, sql: str) -> None:
     """
     Execute DDL only if we are running on Microsoft SQL Server.
     """
-    DDL(sql, bind=engine).execute_if(dialect=SqlaDialectName.SQLSERVER)
+    ddl = DDL(sql).execute_if(dialect=SqlaDialectName.SQLSERVER)
+    execute_ddl(engine, ddl=ddl)
 
 
 @contextmanager