SQLAlchemy 2.0 Compatibility

In the do_execute function of SQLAlchemy dialect, an if statement was accessing context.should_autocommit.
This property has been removed in SQLAlchemy 2.0.
To fix, the if statement can simply be removed as it is no longer needed.

SQLAlchemy 2.0 also removed support for parameters as keyword arguments in the Connection.execute function.
Instead, parameters need to be passed in a dictionary.

Author: mathiasritter

.github/workflows/ci.yml

@@ -58,6 +58,8 @@ jobs:
           - { python: "3.11", trino: "351", sqlalchemy: "~=1.4.0" }  # first Trino version
           # Test with sqlalchemy 1.3
           - { python: "3.11", trino: "latest", sqlalchemy: "~=1.3.0" }
+          # Test with sqlalchemy 2.0
+          - { python: "3.11", trino: "latest", sqlalchemy: "~=2.0.0rc1" }
     env:
       TRINO_VERSION: "${{ matrix.trino }}"
     steps:

setup.py

@@ -26,7 +26,7 @@
     version = str(ast.literal_eval(trino_version.group(1)))
 
 kerberos_require = ["requests_kerberos"]
-sqlalchemy_require = ["sqlalchemy~=1.3"]
+sqlalchemy_require = ["sqlalchemy >= 1.3"]
 external_authentication_token_cache_require = ["keyring"]
 
 # We don't add localstorage_require to all_require as users must explicitly opt in to use keyring.
@@ -78,7 +78,7 @@
         "Programming Language :: Python :: Implementation :: PyPy",
         "Topic :: Database :: Front-Ends",
     ],
-    python_requires='>=3.7',
+    python_requires=">=3.7",
     install_requires=["pytz", "requests", "tzlocal"],
     extras_require={
         "all": all_require,

trino/sqlalchemy/dialect.py

@@ -156,7 +156,7 @@ def _get_columns(self, connection: Connection, table_name: str, schema: str = No
             ORDER BY "ordinal_position" ASC
         """
         ).strip()
-        res = connection.execute(sql.text(query), schema=schema, table=table_name)
+        res = connection.execute(sql.text(query), { "schema": schema, "table": table_name })
         columns = []
         for record in res:
             column = dict(
@@ -204,7 +204,7 @@ def get_table_names(self, connection: Connection, schema: str = None, **kw) -> L
               AND "table_type" = 'BASE TABLE'
         """
         ).strip()
-        res = connection.execute(sql.text(query), schema=schema)
+        res = connection.execute(sql.text(query), { "schema": schema })
         return [row.table_name for row in res]
 
     def get_temp_table_names(self, connection: Connection, schema: str = None, **kw) -> List[str]:
@@ -225,7 +225,7 @@ def get_view_names(self, connection: Connection, schema: str = None, **kw) -> Li
               AND "table_type" = 'VIEW'
         """
         ).strip()
-        res = connection.execute(sql.text(query), schema=schema)
+        res = connection.execute(sql.text(query), { "schema": schema })
         return [row.table_name for row in res]
 
     def get_temp_view_names(self, connection: Connection, schema: str = None, **kw) -> List[str]:
@@ -244,7 +244,7 @@ def get_view_definition(self, connection: Connection, view_name: str, schema: st
               AND "table_name" = :view
         """
         ).strip()
-        res = connection.execute(sql.text(query), schema=schema, view=view_name)
+        res = connection.execute(sql.text(query), { "schema": schema, "view": view_name })
         return res.scalar()
 
     def get_indexes(self, connection: Connection, table_name: str, schema: str = None, **kw) -> List[Dict[str, Any]]:
@@ -314,7 +314,7 @@ def has_schema(self, connection: Connection, schema: str) -> bool:
             WHERE "schema_name" = :schema
         """
         ).strip()
-        res = connection.execute(sql.text(query), schema=schema)
+        res = connection.execute(sql.text(query), { "schema": schema })
         return res.first() is not None
 
     def has_table(self, connection: Connection, table_name: str, schema: str = None, **kw) -> bool:
@@ -329,7 +329,7 @@ def has_table(self, connection: Connection, table_name: str, schema: str = None,
               AND "table_name" = :table
         """
         ).strip()
-        res = connection.execute(sql.text(query), schema=schema, table=table_name)
+        res = connection.execute(sql.text(query), { "schema": schema, "table": table_name })
         return res.first() is not None
 
     def has_sequence(self, connection: Connection, sequence_name: str, schema: str = None, **kw) -> bool:
@@ -363,11 +363,6 @@ def do_execute(
         self, cursor: Cursor, statement: str, parameters: Tuple[Any, ...], context: DefaultExecutionContext = None
     ):
         cursor.execute(statement, parameters)
-        if context and context.should_autocommit:
-            # SQL statement only submitted to Trino server when cursor.fetch*() is called.
-            # For DDL (CREATE/ALTER/DROP) and DML (INSERT/UPDATE/DELETE) statement, call cursor.description
-            # to force submit statement immediately.
-            cursor.description  # noqa
 
     def do_rollback(self, dbapi_connection: trino_dbapi.Connection):
         if dbapi_connection.transaction is not None: