Deprecate plain string in execute and introduce `exec_driver_sql`

Execution of literal sql string is deprecated in the
:meth:`.Connection.execute` and a warning is raised when used stating
that it will be coerced to :func:`.text` in a future release.
To execute a raw sql string the new connection method
:meth:`.Connection.exec_driver_sql` was added, that will retain the previous
behavior, passing the string to the DBAPI driver unchanged.
Usage of scalar or tuple positional parameters in :meth:`.Connection.execute`
is also deprecated.

Fixes: #4848
Fixes: #5178
Change-Id: I2830181054327996d594f7f0d59c157d477c3aa9

15 files changed, 433 insertions, 327 deletions

diff --git a/test/dialect/mssql/test_engine.py b/test/dialect/mssql/test_engine.py
index 257e41bf8..97e924fed 100644
--- a/test/dialect/mssql/test_engine.py
+++ b/test/dialect/mssql/test_engine.py
@@ -400,7 +400,15 @@ class FastExecutemanyTest(fixtures.TestBase):
 
 
 class VersionDetectionTest(fixtures.TestBase):
-    def test_pymssql_version(self):
+    @testing.fixture
+    def mock_conn_scalar(self):
+        return lambda text: Mock(
+            exec_driver_sql=Mock(
+                return_value=Mock(scalar=Mock(return_value=text))
+            )
+        )
+
+    def test_pymssql_version(self, mock_conn_scalar):
         dialect = pymssql.MSDialect_pymssql()
 
         for vers in [
@@ -410,13 +418,13 @@ class VersionDetectionTest(fixtures.TestBase):
             "Microsoft SQL Azure (RTM) - 11.0.9216.62 \n"
             "Jul 18 2014 22:00:21 \nCopyright (c) Microsoft Corporation",
         ]:
-            conn = Mock(scalar=Mock(return_value=vers))
+            conn = mock_conn_scalar(vers)
             eq_(dialect._get_server_version_info(conn), (11, 0, 9216, 62))
 
-    def test_pyodbc_version_productversion(self):
+    def test_pyodbc_version_productversion(self, mock_conn_scalar):
         dialect = pyodbc.MSDialect_pyodbc()
 
-        conn = Mock(scalar=Mock(return_value="11.0.9216.62"))
+        conn = mock_conn_scalar("11.0.9216.62")
         eq_(dialect._get_server_version_info(conn), (11, 0, 9216, 62))
 
     def test_pyodbc_version_fallback(self):
@@ -429,8 +437,12 @@ class VersionDetectionTest(fixtures.TestBase):
             ("Not SQL Server Version 10.5", (5,)),
         ]:
             conn = Mock(
-                scalar=Mock(
-                    side_effect=exc.DBAPIError("stmt", "params", None)
+                exec_driver_sql=Mock(
+                    return_value=Mock(
+                        scalar=Mock(
+                            side_effect=exc.DBAPIError("stmt", "params", None)
+                        )
+                    )
                 ),
                 connection=Mock(getinfo=Mock(return_value=vers)),
             )
@@ -462,7 +474,7 @@ class RealIsolationLevelTest(fixtures.TestBase):
             with testing.db.connect() as c:
                 c.execution_options(isolation_level=value)
 
-                c.execute("SELECT TOP 10 * FROM test")
+                c.exec_driver_sql("SELECT TOP 10 * FROM test")
 
                 eq_(
                     testing.db.dialect.get_isolation_level(c.connection), value
diff --git a/test/dialect/mssql/test_query.py b/test/dialect/mssql/test_query.py
index fc3352d4e..2ad6c161d 100644
--- a/test/dialect/mssql/test_query.py
+++ b/test/dialect/mssql/test_query.py
@@ -325,7 +325,7 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
         )
         meta.create_all()
         con = testing.db.connect()
-        con.execute(
+        con.exec_driver_sql(
             """create trigger paj on t1 for insert as
             insert into t2 (descr) select descr from inserted"""
         )
@@ -339,7 +339,7 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
 
         finally:
             tr.commit()
-            con.execute("""drop trigger paj""")
+            con.exec_driver_sql("""drop trigger paj""")
             meta.drop_all()
 
     @testing.provide_metadata
@@ -431,11 +431,11 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
         )
         meta.bind = eng
         con = eng.connect()
-        con.execute("create schema paj")
+        con.exec_driver_sql("create schema paj")
 
         @event.listens_for(meta, "after_drop")
         def cleanup(target, connection, **kw):
-            connection.execute("drop schema paj")
+            connection.exec_driver_sql("drop schema paj")
 
         tbl = Table(
             "test", meta, Column("id", Integer, primary_key=True), schema="paj"
@@ -450,11 +450,11 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
         eng = engines.testing_engine(options=dict(legacy_schema_aliasing=True))
         meta.bind = eng
         con = eng.connect()
-        con.execute("create schema paj")
+        con.exec_driver_sql("create schema paj")
 
         @event.listens_for(meta, "after_drop")
         def cleanup(target, connection, **kw):
-            connection.execute("drop schema paj")
+            connection.exec_driver_sql("drop schema paj")
 
         tbl = Table(
             "test", meta, Column("id", Integer, primary_key=True), schema="paj"
@@ -489,11 +489,11 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
         )
         meta.bind = eng
         con = eng.connect()
-        con.execute("create schema paj")
+        con.exec_driver_sql("create schema paj")
 
         @event.listens_for(meta, "after_drop")
         def cleanup(target, connection, **kw):
-            connection.execute("drop schema paj")
+            connection.exec_driver_sql("drop schema paj")
 
         tbl = Table(
             "test", meta, Column("id", Integer, primary_key=True), schema="paj"
@@ -510,11 +510,11 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
         eng = engines.testing_engine(options=dict(legacy_schema_aliasing=True))
         meta.bind = eng
         con = eng.connect()
-        con.execute("create schema paj")
+        con.exec_driver_sql("create schema paj")
 
         @event.listens_for(meta, "after_drop")
         def cleanup(target, connection, **kw):
-            connection.execute("drop schema paj")
+            connection.exec_driver_sql("drop schema paj")
 
         tbl = Table(
             "test", meta, Column("id", Integer, primary_key=True), schema="paj"
@@ -548,7 +548,9 @@ def full_text_search_missing():
     try:
         connection = testing.db.connect()
         try:
-            connection.execute("CREATE FULLTEXT CATALOG Catalog AS " "DEFAULT")
+            connection.exec_driver_sql(
+                "CREATE FULLTEXT CATALOG Catalog AS " "DEFAULT"
+            )
             return False
         except Exception:
             return True
@@ -621,7 +623,7 @@ class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
     def teardown_class(cls):
         metadata.drop_all()
         connection = testing.db.connect()
-        connection.execute("DROP FULLTEXT CATALOG Catalog")
+        connection.exec_driver_sql("DROP FULLTEXT CATALOG Catalog")
         connection.close()
 
     def test_expression(self):
diff --git a/test/dialect/mssql/test_reflection.py b/test/dialect/mssql/test_reflection.py
index 352bc637e..5328513e4 100644
--- a/test/dialect/mssql/test_reflection.py
+++ b/test/dialect/mssql/test_reflection.py
@@ -144,11 +144,10 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
     @testing.provide_metadata
     def test_skip_types(self):
         metadata = self.metadata
-        testing.db.execute(
-            """
-            create table foo (id integer primary key, data xml)
-        """
-        )
+        with testing.db.connect() as c:
+            c.exec_driver_sql(
+                "create table foo (id integer primary key, data xml)"
+            )
         with mock.patch.object(
             testing.db.dialect, "ischema_names", {"int": mssql.INTEGER}
         ):
@@ -236,8 +235,9 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
         )
         metadata.create_all()
 
-        dbname = testing.db.scalar("select db_name()")
-        owner = testing.db.scalar("SELECT user_name()")
+        with testing.db.connect() as c:
+            dbname = c.exec_driver_sql("select db_name()").scalar()
+            owner = c.exec_driver_sql("SELECT user_name()").scalar()
         referred_schema = "%(dbname)s.%(owner)s" % {
             "dbname": dbname,
             "owner": owner,
@@ -439,11 +439,21 @@ class OwnerPlusDBTest(fixtures.TestBase):
         schema, owner = base._owner_plus_db(dialect, identifier)
 
         mock_connection = mock.Mock(
-            dialect=dialect, scalar=mock.Mock(return_value="my_db")
+            dialect=dialect,
+            exec_driver_sql=mock.Mock(
+                return_value=mock.Mock(scalar=mock.Mock(return_value="my_db"))
+            ),
         )
         mock_lambda = mock.Mock()
         base._switch_db(schema, mock_connection, mock_lambda, "x", y="bar")
-        eq_(mock_connection.mock_calls, [mock.call.scalar("select db_name()")])
+        eq_(
+            mock_connection.mock_calls,
+            [mock.call.exec_driver_sql("select db_name()")],
+        )
+        eq_(
+            mock_connection.exec_driver_sql.return_value.mock_calls,
+            [mock.call.scalar()],
+        ),
         eq_(mock_lambda.mock_calls, [mock.call("x", y="bar")])
 
     def test_owner_database_pairs_switch_for_different_db(self):
@@ -453,17 +463,24 @@ class OwnerPlusDBTest(fixtures.TestBase):
         schema, owner = base._owner_plus_db(dialect, identifier)
 
         mock_connection = mock.Mock(
-            dialect=dialect, scalar=mock.Mock(return_value="my_db")
+            dialect=dialect,
+            exec_driver_sql=mock.Mock(
+                return_value=mock.Mock(scalar=mock.Mock(return_value="my_db"))
+            ),
         )
         mock_lambda = mock.Mock()
         base._switch_db(schema, mock_connection, mock_lambda, "x", y="bar")
         eq_(
             mock_connection.mock_calls,
             [
-                mock.call.scalar("select db_name()"),
-                mock.call.execute("use my_other_db"),
-                mock.call.execute("use my_db"),
+                mock.call.exec_driver_sql("select db_name()"),
+                mock.call.exec_driver_sql("use my_other_db"),
+                mock.call.exec_driver_sql("use my_db"),
             ],
+            eq_(
+                mock_connection.exec_driver_sql.return_value.mock_calls,
+                [mock.call.scalar()],
+            ),
         )
         eq_(mock_lambda.mock_calls, [mock.call("x", y="bar")])
 
@@ -496,7 +513,11 @@ class OwnerPlusDBTest(fixtures.TestBase):
 
             mock_connection = mock.Mock(
                 dialect=dialect,
-                scalar=mock.Mock(return_value="Some ] Database"),
+                exec_driver_sql=mock.Mock(
+                    return_value=mock.Mock(
+                        scalar=mock.Mock(return_value="Some ] Database")
+                    )
+                ),
             )
             mock_lambda = mock.Mock()
             base._switch_db(schema, mock_connection, mock_lambda, "x", y="bar")
@@ -506,9 +527,13 @@ class OwnerPlusDBTest(fixtures.TestBase):
                 eq_(
                     mock_connection.mock_calls,
                     [
-                        mock.call.scalar("select db_name()"),
-                        mock.call.execute(use_stmt),
-                        mock.call.execute("use [Some  Database]"),
+                        mock.call.exec_driver_sql("select db_name()"),
+                        mock.call.exec_driver_sql(use_stmt),
+                        mock.call.exec_driver_sql("use [Some  Database]"),
                     ],
                 )
+                eq_(
+                    mock_connection.exec_driver_sql.return_value.mock_calls,
+                    [mock.call.scalar()],
+                )
             eq_(mock_lambda.mock_calls, [mock.call("x", y="bar")])
diff --git a/test/dialect/mssql/test_types.py b/test/dialect/mssql/test_types.py
index ba7ed4667..9904408a3 100644
--- a/test/dialect/mssql/test_types.py
+++ b/test/dialect/mssql/test_types.py
@@ -177,7 +177,7 @@ class RowVersionTest(fixtures.TablesTest):
 
         with testing.db.connect() as conn:
             conn.execute(t.insert().values(data="foo"))
-            last_ts_1 = conn.scalar("SELECT @@DBTS")
+            last_ts_1 = conn.exec_driver_sql("SELECT @@DBTS").scalar()
 
             if convert_int:
                 last_ts_1 = int(codecs.encode(last_ts_1, "hex"), 16)
@@ -187,7 +187,7 @@ class RowVersionTest(fixtures.TablesTest):
             conn.execute(
                 t.update().values(data="bar").where(t.c.data == "foo")
             )
-            last_ts_2 = conn.scalar("SELECT @@DBTS")
+            last_ts_2 = conn.exec_driver_sql("SELECT @@DBTS").scalar()
             if convert_int:
                 last_ts_2 = int(codecs.encode(last_ts_2, "hex"), 16)
 
diff --git a/test/dialect/mysql/test_dialect.py b/test/dialect/mysql/test_dialect.py
index aac1f6489..c641411e1 100644
--- a/test/dialect/mysql/test_dialect.py
+++ b/test/dialect/mysql/test_dialect.py
@@ -184,9 +184,9 @@ class DialectTest(fixtures.TestBase):
                 statement = "SELECT 1 FROM DUAL WHERE 1=0"
             return real_exec(self, statement, *args, **kw)
 
-        real_exec = engine._connection_cls._execute_text
+        real_exec = engine._connection_cls.exec_driver_sql
         with mock.patch.object(
-            engine._connection_cls, "_execute_text", my_execute
+            engine._connection_cls, "exec_driver_sql", my_execute
         ):
             with expect_warnings(
                 "Could not retrieve SQL_MODE; please ensure the "
@@ -198,10 +198,10 @@ class DialectTest(fixtures.TestBase):
         c = testing.db.connect().execution_options(
             isolation_level="AUTOCOMMIT"
         )
-        assert c.execute("SELECT @@autocommit;").scalar()
+        assert c.exec_driver_sql("SELECT @@autocommit;").scalar()
 
         c = c.execution_options(isolation_level="READ COMMITTED")
-        assert not c.execute("SELECT @@autocommit;").scalar()
+        assert not c.exec_driver_sql("SELECT @@autocommit;").scalar()
 
     def test_isolation_level(self):
         values = [
diff --git a/test/dialect/mysql/test_for_update.py b/test/dialect/mysql/test_for_update.py
index 3537c3220..1e8df858f 100644
--- a/test/dialect/mysql/test_for_update.py
+++ b/test/dialect/mysql/test_for_update.py
@@ -60,7 +60,7 @@ class MySQLForUpdateLockingTest(fixtures.DeclarativeMappedTest):
     @contextlib.contextmanager
     def run_test(self):
         connection = testing.db.connect()
-        connection.execute("set innodb_lock_wait_timeout=1")
+        connection.exec_driver_sql("set innodb_lock_wait_timeout=1")
         main_trans = connection.begin()
         try:
             yield Session(bind=connection)
@@ -71,7 +71,7 @@ class MySQLForUpdateLockingTest(fixtures.DeclarativeMappedTest):
     def _assert_a_is_locked(self, should_be_locked):
         A = self.classes.A
         with testing.db.begin() as alt_trans:
-            alt_trans.execute("set innodb_lock_wait_timeout=1")
+            alt_trans.exec_driver_sql("set innodb_lock_wait_timeout=1")
             # set x/y > 10
             try:
                 alt_trans.execute(update(A).values(x=15, y=19))
@@ -84,7 +84,7 @@ class MySQLForUpdateLockingTest(fixtures.DeclarativeMappedTest):
     def _assert_b_is_locked(self, should_be_locked):
         B = self.classes.B
         with testing.db.begin() as alt_trans:
-            alt_trans.execute("set innodb_lock_wait_timeout=1")
+            alt_trans.exec_driver_sql("set innodb_lock_wait_timeout=1")
             # set x/y > 10
             try:
                 alt_trans.execute(update(B).values(x=15, y=19))
diff --git a/test/dialect/mysql/test_reflection.py b/test/dialect/mysql/test_reflection.py
index 1ab646f3c..d5ff1314b 100644
--- a/test/dialect/mysql/test_reflection.py
+++ b/test/dialect/mysql/test_reflection.py
@@ -493,12 +493,14 @@ class ReflectionTest(fixtures.TestBase, AssertsCompiledSQL):
         self.metadata.create_all()
 
         with testing.db.connect() as conn:
-            conn.execute("CREATE VIEW v1 AS SELECT * FROM x")
-            conn.execute("CREATE ALGORITHM=MERGE VIEW v2 AS SELECT * FROM x")
-            conn.execute(
+            conn.exec_driver_sql("CREATE VIEW v1 AS SELECT * FROM x")
+            conn.exec_driver_sql(
+                "CREATE ALGORITHM=MERGE VIEW v2 AS SELECT * FROM x"
+            )
+            conn.exec_driver_sql(
                 "CREATE ALGORITHM=UNDEFINED VIEW v3 AS SELECT * FROM x"
             )
-            conn.execute(
+            conn.exec_driver_sql(
                 "CREATE DEFINER=CURRENT_USER VIEW v4 AS SELECT * FROM x"
             )
 
@@ -506,7 +508,7 @@ class ReflectionTest(fixtures.TestBase, AssertsCompiledSQL):
         def cleanup(*arg, **kw):
             with testing.db.connect() as conn:
                 for v in ["v1", "v2", "v3", "v4"]:
-                    conn.execute("DROP VIEW %s" % v)
+                    conn.exec_driver_sql("DROP VIEW %s" % v)
 
         insp = inspect(testing.db)
         for v in ["v1", "v2", "v3", "v4"]:
@@ -523,15 +525,17 @@ class ReflectionTest(fixtures.TestBase, AssertsCompiledSQL):
         @event.listens_for(self.metadata, "before_drop")
         def cleanup(*arg, **kw):
             with testing.db.connect() as conn:
-                conn.execute("DROP TABLE IF EXISTS test_t1")
-                conn.execute("DROP TABLE IF EXISTS test_t2")
-                conn.execute("DROP VIEW IF EXISTS test_v")
+                conn.exec_driver_sql("DROP TABLE IF EXISTS test_t1")
+                conn.exec_driver_sql("DROP TABLE IF EXISTS test_t2")
+                conn.exec_driver_sql("DROP VIEW IF EXISTS test_v")
 
         with testing.db.connect() as conn:
-            conn.execute("CREATE TABLE test_t1 (id INTEGER)")
-            conn.execute("CREATE TABLE test_t2 (id INTEGER)")
-            conn.execute("CREATE VIEW test_v AS SELECT id FROM test_t1")
-            conn.execute("DROP TABLE test_t1")
+            conn.exec_driver_sql("CREATE TABLE test_t1 (id INTEGER)")
+            conn.exec_driver_sql("CREATE TABLE test_t2 (id INTEGER)")
+            conn.exec_driver_sql(
+                "CREATE VIEW test_v AS SELECT id FROM test_t1"
+            )
+            conn.exec_driver_sql("DROP TABLE test_t1")
 
             m = MetaData()
             with expect_warnings(
@@ -567,10 +571,10 @@ class ReflectionTest(fixtures.TestBase, AssertsCompiledSQL):
 
         # this is ideally one table, but older MySQL versions choke
         # on the multiple TIMESTAMP columns
-
-        row = testing.db.execute(
-            "show variables like '%%explicit_defaults_for_timestamp%%'"
-        ).first()
+        with testing.db.connect() as c:
+            row = c.exec_driver_sql(
+                "show variables like '%%explicit_defaults_for_timestamp%%'"
+            ).first()
         explicit_defaults_for_timestamp = row[1].lower() in ("on", "1", "true")
 
         reflected = []
@@ -591,14 +595,15 @@ class ReflectionTest(fixtures.TestBase, AssertsCompiledSQL):
         ):
             Table("nn_t%d" % idx, meta)  # to allow DROP
 
-            testing.db.execute(
-                """
-                CREATE TABLE nn_t%d (
-                    %s
+            with testing.db.connect() as c:
+                c.exec_driver_sql(
+                    """
+                        CREATE TABLE nn_t%d (
+                            %s
+                        )
+                    """
+                    % (idx, ", \n".join(cols))
                 )
-            """
-                % (idx, ", \n".join(cols))
-            )
 
             reflected.extend(
                 {
diff --git a/test/dialect/mysql/test_types.py b/test/dialect/mysql/test_types.py
index ee626b082..010165ac2 100644
--- a/test/dialect/mysql/test_types.py
+++ b/test/dialect/mysql/test_types.py
@@ -697,12 +697,12 @@ class TypeRoundTripTest(fixtures.TestBase, AssertsExecutionResults):
                 return dt
 
         with testing.db.begin() as conn:
-            now = conn.scalar("select now()")
+            now = conn.exec_driver_sql("select now()").scalar()
             conn.execute(ts_table.insert(), {"t1": now, "t2": None})
             conn.execute(ts_table.insert(), {"t1": None, "t2": None})
             conn.execute(ts_table.insert(), {"t2": None})
 
-            new_now = conn.scalar("select now()")
+            new_now = conn.exec_driver_sql("select now()").scalar()
 
             eq_(
                 [
diff --git a/test/dialect/oracle/test_dialect.py b/test/dialect/oracle/test_dialect.py
index 20c6336b8..c2983bfe0 100644
--- a/test/dialect/oracle/test_dialect.py
+++ b/test/dialect/oracle/test_dialect.py
@@ -331,19 +331,20 @@ class OutParamTest(fixtures.TestBase, AssertsExecutionResults):
 
     @classmethod
     def setup_class(cls):
-        testing.db.execute(
-            """
-        create or replace procedure foo(x_in IN number, x_out OUT number,
-        y_out OUT number, z_out OUT varchar) IS
-        retval number;
-        begin
-            retval := 6;
-            x_out := 10;
-            y_out := x_in * 15;
-            z_out := NULL;
-        end;
-        """
-        )
+        with testing.db.connect() as c:
+            c.exec_driver_sql(
+                """
+create or replace procedure foo(x_in IN number, x_out OUT number,
+y_out OUT number, z_out OUT varchar) IS
+retval number;
+begin
+    retval := 6;
+    x_out := 10;
+    y_out := x_in * 15;
+    z_out := NULL;
+end;
+                """
+            )
 
     def test_out_params(self):
         result = testing.db.execute(
@@ -362,7 +363,7 @@ class OutParamTest(fixtures.TestBase, AssertsExecutionResults):
 
     @classmethod
     def teardown_class(cls):
-        testing.db.execute("DROP PROCEDURE foo")
+        testing.db.execute(text("DROP PROCEDURE foo"))
 
 
 class QuotedBindRoundTripTest(fixtures.TestBase):
@@ -511,7 +512,9 @@ class CompatFlagsTest(fixtures.TestBase, AssertsCompiledSQL):
 
         dialect = self._dialect((12, 2, 0))
         conn = mock.Mock(
-            execute=mock.Mock(return_value=mock.Mock(scalar=lambda: "12.2.0"))
+            exec_driver_sql=mock.Mock(
+                return_value=mock.Mock(scalar=lambda: "12.2.0")
+            )
         )
         dialect.initialize(conn)
         eq_(dialect.server_version_info, (12, 2, 0))
@@ -524,7 +527,9 @@ class CompatFlagsTest(fixtures.TestBase, AssertsCompiledSQL):
         dialect = self._dialect((12, 2, 0))
 
         conn = mock.Mock(
-            execute=mock.Mock(return_value=mock.Mock(scalar=lambda: "12.2.0"))
+            exec_driver_sql=mock.Mock(
+                return_value=mock.Mock(scalar=lambda: "12.2.0")
+            )
         )
         dialect.initialize(conn)
         eq_(dialect.server_version_info, (12, 2, 0))
@@ -540,7 +545,7 @@ class CompatFlagsTest(fixtures.TestBase, AssertsCompiledSQL):
         dialect = self._dialect((11, 2, 0))
 
         conn = mock.Mock(
-            execute=mock.Mock(return_value=mock.Mock(scalar="11.0.0"))
+            exec_driver_sql=mock.Mock(return_value=mock.Mock(scalar="11.0.0"))
         )
         dialect.initialize(conn)
         eq_(dialect.server_version_info, (11, 2, 0))
@@ -553,7 +558,9 @@ class CompatFlagsTest(fixtures.TestBase, AssertsCompiledSQL):
         dialect = self._dialect((12, 2, 0))
 
         conn = mock.Mock(
-            execute=mock.Mock(return_value=mock.Mock(scalar=lambda: "11.0.0"))
+            exec_driver_sql=mock.Mock(
+                return_value=mock.Mock(scalar=lambda: "11.0.0")
+            )
         )
         dialect.initialize(conn)
         eq_(dialect.server_version_info, (12, 2, 0))
@@ -571,7 +578,7 @@ class CompatFlagsTest(fixtures.TestBase, AssertsCompiledSQL):
             )
 
         conn = mock.Mock(
-            execute=mock.Mock(return_value=mock.Mock(scalar=c122))
+            exec_driver_sql=mock.Mock(return_value=mock.Mock(scalar=c122))
         )
         dialect.initialize(conn)
         eq_(dialect.server_version_info, (12, 2, 0))
@@ -590,7 +597,7 @@ class CompatFlagsTest(fixtures.TestBase, AssertsCompiledSQL):
             return "12.thisiscrap.0"
 
         conn = mock.Mock(
-            execute=mock.Mock(return_value=mock.Mock(scalar=c122))
+            exec_driver_sql=mock.Mock(return_value=mock.Mock(scalar=c122))
         )
         dialect.initialize(conn)
         eq_(dialect.server_version_info, (12, 2, 0))
@@ -609,12 +616,13 @@ class ExecuteTest(fixtures.TestBase):
     __backend__ = True
 
     def test_basic(self):
-        eq_(
-            testing.db.execute(
-                "/*+ this is a comment */ SELECT 1 FROM " "DUAL"
-            ).fetchall(),
-            [(1,)],
-        )
+        with testing.db.connect() as conn:
+            eq_(
+                conn.exec_driver_sql(
+                    "/*+ this is a comment */ SELECT 1 FROM " "DUAL"
+                ).fetchall(),
+                [(1,)],
+            )
 
     def test_sequences_are_integers(self):
         seq = Sequence("foo_seq")
diff --git a/test/dialect/oracle/test_reflection.py b/test/dialect/oracle/test_reflection.py
index 6359c5c17..3d1361adb 100644
--- a/test/dialect/oracle/test_reflection.py
+++ b/test/dialect/oracle/test_reflection.py
@@ -32,6 +32,11 @@ from sqlalchemy.testing.schema import Column
 from sqlalchemy.testing.schema import Table
 
 
+def exec_sql(engine, sql, *args, **kwargs):
+    with engine.connect() as conn:
+        return conn.exec_driver_sql(sql, *args, **kwargs)
+
+
 class MultiSchemaTest(fixtures.TestBase, AssertsCompiledSQL):
     __only_on__ = "oracle"
     __backend__ = True
@@ -79,7 +84,7 @@ grant references on %(test_schema)s.child to public;
             % {"test_schema": testing.config.test_schema}
         ).split(";"):
             if stmt.strip():
-                testing.db.execute(stmt)
+                exec_sql(testing.db, stmt)
 
     @classmethod
     def teardown_class(cls):
@@ -97,7 +102,7 @@ drop synonym %(test_schema)s.local_table;
             % {"test_schema": testing.config.test_schema}
         ).split(";"):
             if stmt.strip():
-                testing.db.execute(stmt)
+                exec_sql(testing.db, stmt)
 
     @testing.provide_metadata
     def test_create_same_names_explicit_schema(self):
@@ -221,11 +226,12 @@ drop synonym %(test_schema)s.local_table;
         )
 
     def test_reflect_local_to_remote(self):
-        testing.db.execute(
+        exec_sql(
+            testing.db,
             "CREATE TABLE localtable (id INTEGER "
             "PRIMARY KEY, parent_id INTEGER REFERENCES "
             "%(test_schema)s.parent(id))"
-            % {"test_schema": testing.config.test_schema}
+            % {"test_schema": testing.config.test_schema},
         )
         try:
             meta = MetaData(testing.db)
@@ -242,7 +248,7 @@ drop synonym %(test_schema)s.local_table;
                 parent.join(lcl)
             ).execute().fetchall()
         finally:
-            testing.db.execute("DROP TABLE localtable")
+            exec_sql(testing.db, "DROP TABLE localtable")
 
     def test_reflect_alt_owner_implicit(self):
         meta = MetaData(testing.db)
@@ -264,10 +270,11 @@ drop synonym %(test_schema)s.local_table;
         ).execute().fetchall()
 
     def test_reflect_alt_owner_synonyms(self):
-        testing.db.execute(
+        exec_sql(
+            testing.db,
             "CREATE TABLE localtable (id INTEGER "
             "PRIMARY KEY, parent_id INTEGER REFERENCES "
-            "%s.ptable(id))" % testing.config.test_schema
+            "%s.ptable(id))" % testing.config.test_schema,
         )
         try:
             meta = MetaData(testing.db)
@@ -286,7 +293,7 @@ drop synonym %(test_schema)s.local_table;
                 parent.join(lcl)
             ).execute().fetchall()
         finally:
-            testing.db.execute("DROP TABLE localtable")
+            exec_sql(testing.db, "DROP TABLE localtable")
 
     def test_reflect_remote_synonyms(self):
         meta = MetaData(testing.db)
@@ -364,18 +371,19 @@ class SystemTableTablenamesTest(fixtures.TestBase):
     __backend__ = True
 
     def setup(self):
-        testing.db.execute("create table my_table (id integer)")
-        testing.db.execute(
-            "create global temporary table my_temp_table (id integer)"
+        exec_sql(testing.db, "create table my_table (id integer)")
+        exec_sql(
+            testing.db,
+            "create global temporary table my_temp_table (id integer)",
         )
-        testing.db.execute(
-            "create table foo_table (id integer) tablespace SYSTEM"
+        exec_sql(
+            testing.db, "create table foo_table (id integer) tablespace SYSTEM"
         )
 
     def teardown(self):
-        testing.db.execute("drop table my_temp_table")
-        testing.db.execute("drop table my_table")
-        testing.db.execute("drop table foo_table")
+        exec_sql(testing.db, "drop table my_temp_table")
+        exec_sql(testing.db, "drop table my_table")
+        exec_sql(testing.db, "drop table foo_table")
 
     def test_table_names_no_system(self):
         insp = inspect(testing.db)
@@ -404,7 +412,8 @@ class DontReflectIOTTest(fixtures.TestBase):
     __backend__ = True
 
     def setup(self):
-        testing.db.execute(
+        exec_sql(
+            testing.db,
             """
         CREATE TABLE admin_docindex(
                 token char(20),
@@ -416,11 +425,11 @@ class DontReflectIOTTest(fixtures.TestBase):
             TABLESPACE users
             PCTTHRESHOLD 20
             OVERFLOW TABLESPACE users
-        """
+        """,
         )
 
     def teardown(self):
-        testing.db.execute("drop table admin_docindex")
+        exec_sql(testing.db, "drop table admin_docindex")
 
     def test_reflect_all(self):
         m = MetaData(testing.db)
@@ -443,8 +452,9 @@ class UnsupportedIndexReflectTest(fixtures.TestBase):
         )
         metadata.create_all()
 
-        testing.db.execute(
-            "CREATE INDEX DATA_IDX ON " "TEST_INDEX_REFLECT (UPPER(DATA))"
+        exec_sql(
+            testing.db,
+            "CREATE INDEX DATA_IDX ON " "TEST_INDEX_REFLECT (UPPER(DATA))",
         )
         m2 = MetaData(testing.db)
         Table("test_index_reflect", m2, autoload=True)
@@ -452,9 +462,10 @@ class UnsupportedIndexReflectTest(fixtures.TestBase):
 
 def all_tables_compression_missing():
     try:
-        testing.db.execute("SELECT compression FROM all_tables")
-        if "Enterprise Edition" not in testing.db.scalar(
-            "select * from v$version"
+        exec_sql(testing.db, "SELECT compression FROM all_tables")
+        if (
+            "Enterprise Edition"
+            not in exec_sql(testing.db, "select * from v$version").scalar()
         ):
             return True
         return False
@@ -464,9 +475,10 @@ def all_tables_compression_missing():
 
 def all_tables_compress_for_missing():
     try:
-        testing.db.execute("SELECT compress_for FROM all_tables")
-        if "Enterprise Edition" not in testing.db.scalar(
-            "select * from v$version"
+        exec_sql(testing.db, "SELECT compress_for FROM all_tables")
+        if (
+            "Enterprise Edition"
+            not in exec_sql(testing.db, "select * from v$version").scalar()
         ):
             return True
         return False
@@ -628,11 +640,11 @@ class DBLinkReflectionTest(fixtures.TestBase):
         # when accessing via a different username as we do with the
         # multiprocess test suite, so testing here is minimal
         with testing.db.connect() as conn:
-            conn.execute(
+            conn.exec_driver_sql(
                 "create table test_table "
                 "(id integer primary key, data varchar2(50))"
             )
-            conn.execute(
+            conn.exec_driver_sql(
                 "create synonym test_table_syn "
                 "for test_table@%s" % cls.dblink
             )
@@ -640,8 +652,8 @@ class DBLinkReflectionTest(fixtures.TestBase):
     @classmethod
     def teardown_class(cls):
         with testing.db.connect() as conn:
-            conn.execute("drop synonym test_table_syn")
-            conn.execute("drop table test_table")
+            conn.exec_driver_sql("drop synonym test_table_syn")
+            conn.exec_driver_sql("drop table test_table")
 
     def test_reflection(self):
         """test the resolution of the synonym/dblink. """
diff --git a/test/dialect/oracle/test_types.py b/test/dialect/oracle/test_types.py
index 23c73a231..70c8f20f2 100644
--- a/test/dialect/oracle/test_types.py
+++ b/test/dialect/oracle/test_types.py
@@ -51,6 +51,11 @@ from sqlalchemy.util import py2k
 from sqlalchemy.util import u
 
 
+def exec_sql(engine, sql, *args, **kwargs):
+    with engine.connect() as conn:
+        return conn.exec_driver_sql(sql, *args, **kwargs)
+
+
 class DialectTypesTest(fixtures.TestBase, AssertsCompiledSQL):
     __dialect__ = oracle.OracleDialect()
 
@@ -371,8 +376,8 @@ class TypesTest(fixtures.TestBase):
         )
 
         eq_(
-            testing.db.execute(
-                "select numericcol from t1 order by intcol"
+            exec_sql(
+                testing.db, "select numericcol from t1 order by intcol"
             ).fetchall(),
             [(float("inf"),), (float("-inf"),)],
         )
@@ -403,8 +408,8 @@ class TypesTest(fixtures.TestBase):
         )
 
         eq_(
-            testing.db.execute(
-                "select numericcol from t1 order by intcol"
+            exec_sql(
+                testing.db, "select numericcol from t1 order by intcol"
             ).fetchall(),
             [(decimal.Decimal("Infinity"),), (decimal.Decimal("-Infinity"),)],
         )
@@ -439,8 +444,8 @@ class TypesTest(fixtures.TestBase):
         eq_(
             [
                 tuple(str(col) for col in row)
-                for row in testing.db.execute(
-                    "select numericcol from t1 order by intcol"
+                for row in exec_sql(
+                    testing.db, "select numericcol from t1 order by intcol"
                 )
             ],
             [("nan",), ("nan",)],
@@ -474,8 +479,8 @@ class TypesTest(fixtures.TestBase):
         )
 
         eq_(
-            testing.db.execute(
-                "select numericcol from t1 order by intcol"
+            exec_sql(
+                testing.db, "select numericcol from t1 order by intcol"
             ).fetchall(),
             [(decimal.Decimal("NaN"),), (decimal.Decimal("NaN"),)],
         )
@@ -515,7 +520,7 @@ class TypesTest(fixtures.TestBase):
 
         stmt = "SELECT idata, ndata, ndata2, nidata, fdata FROM foo"
 
-        row = testing.db.execute(stmt).fetchall()[0]
+        row = exec_sql(testing.db, stmt).fetchall()[0]
         eq_(
             [type(x) for x in row],
             [int, decimal.Decimal, decimal.Decimal, int, float],
@@ -551,7 +556,7 @@ class TypesTest(fixtures.TestBase):
             (SELECT CAST((SELECT fdata FROM foo) AS FLOAT) FROM DUAL) AS fdata
         FROM dual
         """
-        row = testing.db.execute(stmt).fetchall()[0]
+        row = exec_sql(testing.db, stmt).fetchall()[0]
         eq_(
             [type(x) for x in row],
             [int, decimal.Decimal, int, int, decimal.Decimal],
@@ -608,7 +613,7 @@ class TypesTest(fixtures.TestBase):
         )
         WHERE ROWNUM >= 0) anon_1
         """
-        row = testing.db.execute(stmt).fetchall()[0]
+        row = exec_sql(testing.db, stmt).fetchall()[0]
         eq_(
             [type(x) for x in row],
             [int, decimal.Decimal, int, int, decimal.Decimal],
@@ -660,7 +665,7 @@ class TypesTest(fixtures.TestBase):
         engine = testing_engine(options=dict(coerce_to_decimal=False))
 
         # raw SQL no longer coerces to decimal
-        value = engine.scalar("SELECT 5.66 FROM DUAL")
+        value = exec_sql(engine, "SELECT 5.66 FROM DUAL").scalar()
         assert isinstance(value, float)
 
         # explicit typing still *does* coerce to decimal
@@ -673,7 +678,7 @@ class TypesTest(fixtures.TestBase):
         assert isinstance(value, decimal.Decimal)
 
         # default behavior is raw SQL coerces to decimal
-        value = testing.db.scalar("SELECT 5.66 FROM DUAL")
+        value = exec_sql(testing.db, "SELECT 5.66 FROM DUAL").scalar()
         assert isinstance(value, decimal.Decimal)
 
     @testing.combinations(
@@ -713,7 +718,7 @@ class TypesTest(fixtures.TestBase):
             cx_oracle_result = cursor.fetchone()[0]
             cursor.close()
 
-            sqla_result = conn.scalar(stmt)
+            sqla_result = conn.exec_driver_sql(stmt).scalar()
 
             eq_(sqla_result, cx_oracle_result)
 
@@ -723,10 +728,10 @@ class TypesTest(fixtures.TestBase):
     )
     def test_coerce_to_unicode(self):
         engine = testing_engine(options=dict(coerce_to_unicode=False))
-        value = engine.scalar("SELECT 'hello' FROM DUAL")
+        value = exec_sql(engine, "SELECT 'hello' FROM DUAL").scalar()
         assert isinstance(value, util.binary_type)
 
-        value = testing.db.scalar("SELECT 'hello' FROM DUAL")
+        value = exec_sql(testing.db, "SELECT 'hello' FROM DUAL").scalar()
         assert isinstance(value, util.text_type)
 
     @testing.provide_metadata
@@ -865,21 +870,22 @@ class TypesTest(fixtures.TestBase):
 
     def test_longstring(self):
         metadata = MetaData(testing.db)
-        testing.db.execute(
+        exec_sql(
+            testing.db,
             """
         CREATE TABLE Z_TEST
         (
           ID        NUMERIC(22) PRIMARY KEY,
           ADD_USER  VARCHAR2(20)  NOT NULL
         )
-        """
+        """,
         )
         try:
             t = Table("z_test", metadata, autoload=True)
             t.insert().execute(id=1.0, add_user="foobar")
             assert t.select().execute().fetchall() == [(1, "foobar")]
         finally:
-            testing.db.execute("DROP TABLE Z_TEST")
+            exec_sql(testing.db, "DROP TABLE Z_TEST")
 
 
 class LOBFetchTest(fixtures.TablesTest):
@@ -943,7 +949,7 @@ class LOBFetchTest(fixtures.TablesTest):
         eq_(row["bindata"], b("this is binary 1"))
 
     def test_lobs_with_convert_raw(self):
-        row = testing.db.execute("select data, bindata from z_test").first()
+        row = exec_sql(testing.db, "select data, bindata from z_test").first()
         eq_(row["data"], "this is text 1")
         eq_(row["bindata"], b("this is binary 1"))
 
@@ -951,8 +957,8 @@ class LOBFetchTest(fixtures.TablesTest):
         engine = testing_engine(
             options=dict(auto_convert_lobs=False, arraysize=1)
         )
-        result = engine.execute(
-            "select id, data, bindata from z_test order by id"
+        result = exec_sql(
+            engine, "select id, data, bindata from z_test order by id"
         )
         results = result.fetchall()
 
@@ -985,8 +991,8 @@ class LOBFetchTest(fixtures.TablesTest):
         engine = testing_engine(
             options=dict(auto_convert_lobs=True, arraysize=1)
         )
-        result = engine.execute(
-            "select id, data, bindata from z_test order by id"
+        result = exec_sql(
+            engine, "select id, data, bindata from z_test order by id"
         )
         results = result.fetchall()
 
@@ -1090,7 +1096,10 @@ class EuroNumericTest(fixtures.TestBase):
                     {},
                 ),
             ]:
-                test_exp = conn.scalar(stmt, **kw)
+                if isinstance(stmt, util.string_types):
+                    test_exp = conn.exec_driver_sql(stmt, kw).scalar()
+                else:
+                    test_exp = conn.scalar(stmt, **kw)
                 eq_(test_exp, exp)
                 assert type(test_exp) is type(exp)
 
diff --git a/test/dialect/postgresql/test_dialect.py b/test/dialect/postgresql/test_dialect.py
index e36b69802..03e91482d 100644
--- a/test/dialect/postgresql/test_dialect.py
+++ b/test/dialect/postgresql/test_dialect.py
@@ -56,7 +56,7 @@ class DialectTest(fixtures.TestBase):
     def test_version_parsing(self):
         def mock_conn(res):
             return mock.Mock(
-                execute=mock.Mock(
+                exec_driver_sql=mock.Mock(
                     return_value=mock.Mock(scalar=mock.Mock(return_value=res))
                 )
             )
@@ -615,7 +615,7 @@ class MiscBackendTest(
             conn = testing.db.connect()
             trans = conn.begin()
             try:
-                conn.execute(
+                conn.exec_driver_sql(
                     """
 CREATE OR REPLACE FUNCTION note(message varchar) RETURNS integer AS $$
 BEGIN
@@ -625,8 +625,8 @@ END;
 $$ LANGUAGE plpgsql;
 """
                 )
-                conn.execute("SELECT note('hi there')")
-                conn.execute("SELECT note('another note')")
+                conn.exec_driver_sql("SELECT note('hi there')")
+                conn.exec_driver_sql("SELECT note('another note')")
             finally:
                 trans.rollback()
         finally:
@@ -643,7 +643,9 @@ $$ LANGUAGE plpgsql;
     @engines.close_open_connections
     def test_client_encoding(self):
         c = testing.db.connect()
-        current_encoding = c.execute("show client_encoding").fetchone()[0]
+        current_encoding = c.exec_driver_sql(
+            "show client_encoding"
+        ).fetchone()[0]
         c.close()
 
         # attempt to use an encoding that's not
@@ -655,7 +657,7 @@ $$ LANGUAGE plpgsql;
 
         e = engines.testing_engine(options={"client_encoding": test_encoding})
         c = e.connect()
-        new_encoding = c.execute("show client_encoding").fetchone()[0]
+        new_encoding = c.exec_driver_sql("show client_encoding").fetchone()[0]
         eq_(new_encoding, test_encoding)
 
     @testing.requires.psycopg2_or_pg8000_compatibility
@@ -671,7 +673,7 @@ $$ LANGUAGE plpgsql;
         assert_raises_message(
             exc.ProgrammingError,
             'prepared transaction with identifier "gilberte" does not exist',
-            c.execute,
+            c.exec_driver_sql,
             "commit prepared 'gilberte'",
         )
 
@@ -697,7 +699,7 @@ $$ LANGUAGE plpgsql;
         seq = Sequence("fooseq")
         t = Table("mytable", meta1, Column("col1", Integer, seq))
         seq.drop()
-        testing.db.execute("CREATE SEQUENCE fooseq")
+        testing.db.execute(text("CREATE SEQUENCE fooseq"))
         t.create(checkfirst=True)
 
     @testing.provide_metadata
@@ -766,7 +768,8 @@ $$ LANGUAGE plpgsql;
         try:
             meta = MetaData(testing.db)
             testing.db.execute(
-                """
+                text(
+                    """
              CREATE TABLE speedy_users
              (
                  speedy_user_id   SERIAL     PRIMARY KEY,
@@ -775,6 +778,7 @@ $$ LANGUAGE plpgsql;
                  user_password    VARCHAR    NOT NULL
              );
             """
+                )
             )
             t = Table("speedy_users", meta, autoload=True)
             r = t.insert().execute(user_name="user", user_password="lala")
@@ -782,7 +786,7 @@ $$ LANGUAGE plpgsql;
             result = t.select().execute().fetchall()
             assert result == [(1, "user", "lala")]
         finally:
-            testing.db.execute("drop table speedy_users")
+            testing.db.execute(text("drop table speedy_users"))
 
     @testing.requires.psycopg2_or_pg8000_compatibility
     def test_numeric_raise(self):
diff --git a/test/dialect/postgresql/test_reflection.py b/test/dialect/postgresql/test_reflection.py
index 8e26d5a83..89d4ae081 100644
--- a/test/dialect/postgresql/test_reflection.py
+++ b/test/dialect/postgresql/test_reflection.py
@@ -277,28 +277,32 @@ class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults):
             'CREATE DOMAIN "SomeSchema"."Quoted.Domain" INTEGER DEFAULT 0',
         ]:
             try:
-                con.execute(ddl)
+                con.exec_driver_sql(ddl)
             except exc.DBAPIError as e:
                 if "already exists" not in str(e):
                     raise e
-        con.execute(
+        con.exec_driver_sql(
             "CREATE TABLE testtable (question integer, answer " "testdomain)"
         )
-        con.execute(
+        con.exec_driver_sql(
             "CREATE TABLE test_schema.testtable(question "
             "integer, answer test_schema.testdomain, anything "
             "integer)"
         )
-        con.execute(
+        con.exec_driver_sql(
             "CREATE TABLE crosschema (question integer, answer "
             "test_schema.testdomain)"
         )
 
-        con.execute("CREATE TABLE enum_test (id integer, data enumdomain)")
+        con.exec_driver_sql(
+            "CREATE TABLE enum_test (id integer, data enumdomain)"
+        )
 
-        con.execute("CREATE TABLE array_test (id integer, data arraydomain)")
+        con.exec_driver_sql(
+            "CREATE TABLE array_test (id integer, data arraydomain)"
+        )
 
-        con.execute(
+        con.exec_driver_sql(
             "CREATE TABLE quote_test "
             '(id integer, data "SomeSchema"."Quoted.Domain")'
         )
@@ -306,19 +310,19 @@ class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults):
     @classmethod
     def teardown_class(cls):
         con = testing.db.connect()
-        con.execute("DROP TABLE testtable")
-        con.execute("DROP TABLE test_schema.testtable")
-        con.execute("DROP TABLE crosschema")
-        con.execute("DROP TABLE quote_test")
-        con.execute("DROP DOMAIN testdomain")
-        con.execute("DROP DOMAIN test_schema.testdomain")
-        con.execute("DROP TABLE enum_test")
-        con.execute("DROP DOMAIN enumdomain")
-        con.execute("DROP TYPE testtype")
-        con.execute("DROP TABLE array_test")
-        con.execute("DROP DOMAIN arraydomain")
-        con.execute('DROP DOMAIN "SomeSchema"."Quoted.Domain"')
-        con.execute('DROP SCHEMA "SomeSchema"')
+        con.exec_driver_sql("DROP TABLE testtable")
+        con.exec_driver_sql("DROP TABLE test_schema.testtable")
+        con.exec_driver_sql("DROP TABLE crosschema")
+        con.exec_driver_sql("DROP TABLE quote_test")
+        con.exec_driver_sql("DROP DOMAIN testdomain")
+        con.exec_driver_sql("DROP DOMAIN test_schema.testdomain")
+        con.exec_driver_sql("DROP TABLE enum_test")
+        con.exec_driver_sql("DROP DOMAIN enumdomain")
+        con.exec_driver_sql("DROP TYPE testtype")
+        con.exec_driver_sql("DROP TABLE array_test")
+        con.exec_driver_sql("DROP DOMAIN arraydomain")
+        con.exec_driver_sql('DROP DOMAIN "SomeSchema"."Quoted.Domain"')
+        con.exec_driver_sql('DROP SCHEMA "SomeSchema"')
 
     def test_table_is_reflected(self):
         metadata = MetaData(testing.db)
@@ -485,9 +489,9 @@ class ReflectionTest(fixtures.TestBase):
         eq_(t2.c.id.server_default.arg.text, "nextval('t_id_seq'::regclass)")
         r = t2.insert().execute()
         eq_(r.inserted_primary_key, [1])
-        testing.db.connect().execution_options(autocommit=True).execute(
-            "alter table t_id_seq rename to foobar_id_seq"
-        )
+        testing.db.connect().execution_options(
+            autocommit=True
+        ).exec_driver_sql("alter table t_id_seq rename to foobar_id_seq")
         m3 = MetaData(testing.db)
         t3 = Table("t", m3, autoload=True, implicit_returning=False)
         eq_(
@@ -507,9 +511,9 @@ class ReflectionTest(fixtures.TestBase):
             Column("x", Integer),
         )
         metadata.create_all()
-        testing.db.connect().execution_options(autocommit=True).execute(
-            "alter table t alter column id type varchar(50)"
-        )
+        testing.db.connect().execution_options(
+            autocommit=True
+        ).exec_driver_sql("alter table t alter column id type varchar(50)")
         m2 = MetaData(testing.db)
         t2 = Table("t", m2, autoload=True)
         eq_(t2.c.id.autoincrement, False)
@@ -520,9 +524,9 @@ class ReflectionTest(fixtures.TestBase):
         metadata = self.metadata
         Table("t", metadata, Column("id", Integer, primary_key=True))
         metadata.create_all()
-        testing.db.connect().execution_options(autocommit=True).execute(
-            "alter table t rename id to t_id"
-        )
+        testing.db.connect().execution_options(
+            autocommit=True
+        ).exec_driver_sql("alter table t rename id to t_id")
         m2 = MetaData(testing.db)
         t2 = Table("t", m2, autoload=True)
         eq_([c.name for c in t2.primary_key], ["t_id"])
@@ -642,7 +646,7 @@ class ReflectionTest(fixtures.TestBase):
 
         conn = testing.db.connect()
         conn.detach()
-        conn.execute("SET search_path TO test_schema, test_schema_2")
+        conn.exec_driver_sql("SET search_path TO test_schema, test_schema_2")
         meta2 = MetaData(bind=conn)
         subject = Table(
             "subject",
@@ -727,7 +731,7 @@ class ReflectionTest(fixtures.TestBase):
         with testing.db.connect() as conn:
             conn.detach()
 
-            conn.execute(
+            conn.exec_driver_sql(
                 "set search_path to test_schema_2, test_schema, public"
             )
 
@@ -792,7 +796,7 @@ class ReflectionTest(fixtures.TestBase):
         with testing.db.connect() as conn:
             conn.detach()
 
-            conn.execute(
+            conn.exec_driver_sql(
                 "set search_path to test_schema_2, test_schema, public"
             )
             meta2 = MetaData(conn)
@@ -889,22 +893,17 @@ class ReflectionTest(fixtures.TestBase):
             Column("aname", String(20)),
         )
         metadata.create_all()
-        testing.db.execute(
-            """
-          create index idx1 on party ((id || name))
-        """
-        )
-        testing.db.execute(
-            """
-          create unique index idx2 on party (id) where name = 'test'
-        """
-        )
-        testing.db.execute(
-            """
-            create index idx3 on party using btree
-                (lower(name::text), lower(aname::text))
-        """
-        )
+        with testing.db.connect() as c:
+            c.exec_driver_sql("create index idx1 on party ((id || name))")
+            c.exec_driver_sql(
+                "create unique index idx2 on party (id) where name = 'test'"
+            )
+            c.exec_driver_sql(
+                """
+                create index idx3 on party using btree
+                    (lower(name::text), lower(aname::text))
+                """
+            )
 
         def go():
             m2 = MetaData(testing.db)
@@ -951,7 +950,7 @@ class ReflectionTest(fixtures.TestBase):
             t1.create(conn)
 
             # check ASC, DESC options alone
-            conn.execute(
+            conn.exec_driver_sql(
                 """
                 create index idx1 on party
                     (id, name ASC, aname DESC)
@@ -959,7 +958,7 @@ class ReflectionTest(fixtures.TestBase):
             )
 
             # check DESC w/ NULLS options
-            conn.execute(
+            conn.exec_driver_sql(
                 """
               create index idx2 on party
                     (name DESC NULLS FIRST, aname DESC NULLS LAST)
@@ -967,7 +966,7 @@ class ReflectionTest(fixtures.TestBase):
             )
 
             # check ASC w/ NULLS options
-            conn.execute(
+            conn.exec_driver_sql(
                 """
               create index idx3 on party
                     (name ASC NULLS FIRST, aname ASC NULLS LAST)
@@ -1028,8 +1027,8 @@ class ReflectionTest(fixtures.TestBase):
         )
         metadata.create_all()
         conn = testing.db.connect().execution_options(autocommit=True)
-        conn.execute("CREATE INDEX idx1 ON t (x)")
-        conn.execute("ALTER TABLE t RENAME COLUMN x to y")
+        conn.exec_driver_sql("CREATE INDEX idx1 ON t (x)")
+        conn.exec_driver_sql("ALTER TABLE t RENAME COLUMN x to y")
 
         ind = testing.db.dialect.get_indexes(conn, "t", None)
         eq_(ind, [{"unique": False, "column_names": ["y"], "name": "idx1"}])
@@ -1051,7 +1050,9 @@ class ReflectionTest(fixtures.TestBase):
         metadata.create_all()
 
         with testing.db.connect().execution_options(autocommit=True) as conn:
-            conn.execute("CREATE INDEX idx1 ON t (x) WITH (fillfactor = 50)")
+            conn.exec_driver_sql(
+                "CREATE INDEX idx1 ON t (x) WITH (fillfactor = 50)"
+            )
 
             ind = testing.db.dialect.get_indexes(conn, "t", None)
             eq_(
@@ -1089,7 +1090,7 @@ class ReflectionTest(fixtures.TestBase):
         )
         metadata.create_all()
         with testing.db.connect().execution_options(autocommit=True) as conn:
-            conn.execute("CREATE INDEX idx1 ON t USING gin (x)")
+            conn.exec_driver_sql("CREATE INDEX idx1 ON t USING gin (x)")
 
             ind = testing.db.dialect.get_indexes(conn, "t", None)
             eq_(
diff --git a/test/dialect/test_firebird.py b/test/dialect/test_firebird.py
index 1b72da1bc..3455dca66 100644
--- a/test/dialect/test_firebird.py
+++ b/test/dialect/test_firebird.py
@@ -40,17 +40,21 @@ class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults):
     def setup_class(cls):
         con = testing.db.connect()
         try:
-            con.execute(
+            con.exec_driver_sql(
                 "CREATE DOMAIN int_domain AS INTEGER DEFAULT " "42 NOT NULL"
             )
-            con.execute("CREATE DOMAIN str_domain AS VARCHAR(255)")
-            con.execute("CREATE DOMAIN rem_domain AS BLOB SUB_TYPE TEXT")
-            con.execute("CREATE DOMAIN img_domain AS BLOB SUB_TYPE " "BINARY")
+            con.exec_driver_sql("CREATE DOMAIN str_domain AS VARCHAR(255)")
+            con.exec_driver_sql(
+                "CREATE DOMAIN rem_domain AS BLOB SUB_TYPE TEXT"
+            )
+            con.exec_driver_sql(
+                "CREATE DOMAIN img_domain AS BLOB SUB_TYPE " "BINARY"
+            )
         except ProgrammingError as e:
             if "attempt to store duplicate value" not in str(e):
                 raise e
-        con.execute("""CREATE GENERATOR gen_testtable_id""")
-        con.execute(
+        con.exec_driver_sql("""CREATE GENERATOR gen_testtable_id""")
+        con.exec_driver_sql(
             """CREATE TABLE testtable (question int_domain,
                                    answer str_domain DEFAULT 'no answer',
                                    remark rem_domain DEFAULT '',
@@ -60,12 +64,12 @@ class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults):
                                    dt timestamp,
                                    redundant str_domain DEFAULT NULL)"""
         )
-        con.execute(
+        con.exec_driver_sql(
             "ALTER TABLE testtable "
             "ADD CONSTRAINT testtable_pk PRIMARY KEY "
             "(question)"
         )
-        con.execute(
+        con.exec_driver_sql(
             "CREATE TRIGGER testtable_autoid FOR testtable "
             "   ACTIVE BEFORE INSERT AS"
             "   BEGIN"
@@ -77,12 +81,12 @@ class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults):
     @classmethod
     def teardown_class(cls):
         con = testing.db.connect()
-        con.execute("DROP TABLE testtable")
-        con.execute("DROP DOMAIN int_domain")
-        con.execute("DROP DOMAIN str_domain")
-        con.execute("DROP DOMAIN rem_domain")
-        con.execute("DROP DOMAIN img_domain")
-        con.execute("DROP GENERATOR gen_testtable_id")
+        con.exec_driver_sql("DROP TABLE testtable")
+        con.exec_driver_sql("DROP DOMAIN int_domain")
+        con.exec_driver_sql("DROP DOMAIN str_domain")
+        con.exec_driver_sql("DROP DOMAIN rem_domain")
+        con.exec_driver_sql("DROP DOMAIN img_domain")
+        con.exec_driver_sql("DROP GENERATOR gen_testtable_id")
 
     def test_table_is_reflected(self):
         from sqlalchemy.types import (
@@ -222,29 +226,29 @@ ID DOM_ID /* INTEGER NOT NULL */ default 0 )
     @classmethod
     def setup_class(cls):
         con = testing.db.connect()
-        con.execute(cls.AUTOINC_DM)
-        con.execute(cls.MONEY_DM)
-        con.execute(cls.NOSI_DM)
-        con.execute(cls.RIT_TESORERIA_CAPITOLO_DM)
-        con.execute(cls.DEF_ERROR_TB)
-        con.execute(cls.DEF_ERROR_NODOM_TB)
+        con.exec_driver_sql(cls.AUTOINC_DM)
+        con.exec_driver_sql(cls.MONEY_DM)
+        con.exec_driver_sql(cls.NOSI_DM)
+        con.exec_driver_sql(cls.RIT_TESORERIA_CAPITOLO_DM)
+        con.exec_driver_sql(cls.DEF_ERROR_TB)
+        con.exec_driver_sql(cls.DEF_ERROR_NODOM_TB)
 
-        con.execute(cls.DOM_ID)
-        con.execute(cls.TABLE_A)
-        con.execute(cls.TABLE_B)
+        con.exec_driver_sql(cls.DOM_ID)
+        con.exec_driver_sql(cls.TABLE_A)
+        con.exec_driver_sql(cls.TABLE_B)
 
     @classmethod
     def teardown_class(cls):
         con = testing.db.connect()
-        con.execute("DROP TABLE a")
-        con.execute("DROP TABLE b")
-        con.execute("DROP DOMAIN dom_id")
-        con.execute("DROP TABLE def_error_nodom")
-        con.execute("DROP TABLE def_error")
-        con.execute("DROP DOMAIN rit_tesoreria_capitolo_dm")
-        con.execute("DROP DOMAIN nosi_dm")
-        con.execute("DROP DOMAIN money_dm")
-        con.execute("DROP DOMAIN autoinc_dm")
+        con.exec_driver_sql("DROP TABLE a")
+        con.exec_driver_sql("DROP TABLE b")
+        con.exec_driver_sql("DROP DOMAIN dom_id")
+        con.exec_driver_sql("DROP TABLE def_error_nodom")
+        con.exec_driver_sql("DROP TABLE def_error")
+        con.exec_driver_sql("DROP DOMAIN rit_tesoreria_capitolo_dm")
+        con.exec_driver_sql("DROP DOMAIN nosi_dm")
+        con.exec_driver_sql("DROP DOMAIN money_dm")
+        con.exec_driver_sql("DROP DOMAIN autoinc_dm")
 
     def test_tables_are_reflected_same_way(self):
         metadata = MetaData(testing.db)
diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py
index 01ef5f084..ea4cba8cc 100644
--- a/test/dialect/test_sqlite.py
+++ b/test/dialect/test_sqlite.py
@@ -59,6 +59,11 @@ from sqlalchemy.util import u
 from sqlalchemy.util import ue
 
 
+def exec_sql(engine, sql, *args, **kwargs):
+    conn = engine.connect(close_with_result=True)
+    return conn.exec_driver_sql(sql, *args, **kwargs)
+
+
 class TestTypes(fixtures.TestBase, AssertsExecutionResults):
 
     __only_on__ = "sqlite"
@@ -77,23 +82,29 @@ class TestTypes(fixtures.TestBase, AssertsExecutionResults):
         )
         try:
             meta.create_all()
-            testing.db.execute(
-                "INSERT INTO bool_table (id, boo) " "VALUES (1, 'false');"
+            exec_sql(
+                testing.db,
+                "INSERT INTO bool_table (id, boo) " "VALUES (1, 'false');",
             )
-            testing.db.execute(
-                "INSERT INTO bool_table (id, boo) " "VALUES (2, 'true');"
+            exec_sql(
+                testing.db,
+                "INSERT INTO bool_table (id, boo) " "VALUES (2, 'true');",
             )
-            testing.db.execute(
-                "INSERT INTO bool_table (id, boo) " "VALUES (3, '1');"
+            exec_sql(
+                testing.db,
+                "INSERT INTO bool_table (id, boo) " "VALUES (3, '1');",
             )
-            testing.db.execute(
-                "INSERT INTO bool_table (id, boo) " "VALUES (4, '0');"
+            exec_sql(
+                testing.db,
+                "INSERT INTO bool_table (id, boo) " "VALUES (4, '0');",
             )
-            testing.db.execute(
-                "INSERT INTO bool_table (id, boo) " "VALUES (5, 1);"
+            exec_sql(
+                testing.db,
+                "INSERT INTO bool_table (id, boo) " "VALUES (5, 1);",
             )
-            testing.db.execute(
-                "INSERT INTO bool_table (id, boo) " "VALUES (6, 0);"
+            exec_sql(
+                testing.db,
+                "INSERT INTO bool_table (id, boo) " "VALUES (6, 0);",
             )
             eq_(
                 t.select(t.c.boo).order_by(t.c.id).execute().fetchall(),
@@ -176,9 +187,11 @@ class TestTypes(fixtures.TestBase, AssertsExecutionResults):
         testing.db.execute(
             t.insert().values(d=datetime.datetime(2010, 10, 15, 12, 37, 0))
         )
-        testing.db.execute("insert into t (d) values ('2004-05-21T00:00:00')")
+        exec_sql(
+            testing.db, "insert into t (d) values ('2004-05-21T00:00:00')"
+        )
         eq_(
-            testing.db.execute("select * from t order by d").fetchall(),
+            exec_sql(testing.db, "select * from t order by d").fetchall(),
             [("2004-05-21T00:00:00",), ("2010-10-15T12:37:00",)],
         )
         eq_(
@@ -201,9 +214,9 @@ class TestTypes(fixtures.TestBase, AssertsExecutionResults):
         testing.db.execute(
             t.insert().values(d=datetime.datetime(2010, 10, 15, 12, 37, 0))
         )
-        testing.db.execute("insert into t (d) values ('20040521000000')")
+        exec_sql(testing.db, "insert into t (d) values ('20040521000000')")
         eq_(
-            testing.db.execute("select * from t order by d").fetchall(),
+            exec_sql(testing.db, "select * from t order by d").fetchall(),
             [("20040521000000",), ("20101015123700",)],
         )
         eq_(
@@ -223,9 +236,9 @@ class TestTypes(fixtures.TestBase, AssertsExecutionResults):
         t = Table("t", self.metadata, Column("d", sqlite_date))
         self.metadata.create_all(testing.db)
         testing.db.execute(t.insert().values(d=datetime.date(2010, 10, 15)))
-        testing.db.execute("insert into t (d) values ('20040521')")
+        exec_sql(testing.db, "insert into t (d) values ('20040521')")
         eq_(
-            testing.db.execute("select * from t order by d").fetchall(),
+            exec_sql(testing.db, "select * from t order by d").fetchall(),
             [("20040521",), ("20101015",)],
         )
         eq_(
@@ -243,9 +256,9 @@ class TestTypes(fixtures.TestBase, AssertsExecutionResults):
         t = Table("t", self.metadata, Column("d", sqlite_date))
         self.metadata.create_all(testing.db)
         testing.db.execute(t.insert().values(d=datetime.date(2010, 10, 15)))
-        testing.db.execute("insert into t (d) values ('2004|05|21')")
+        exec_sql(testing.db, "insert into t (d) values ('2004|05|21')")
         eq_(
-            testing.db.execute("select * from t order by d").fetchall(),
+            exec_sql(testing.db, "select * from t order by d").fetchall(),
             [("2004|05|21",), ("2010|10|15",)],
         )
         eq_(
@@ -499,12 +512,12 @@ class DefaultsTest(fixtures.TestBase, AssertsCompiledSQL):
             val INTEGER NOT NULL DEFAULT 0
             )"""
         try:
-            db.execute(table)
+            exec_sql(db, table)
             rt = Table("r_defaults", m, autoload=True)
             for i, reflected in enumerate(rt.c):
                 eq_(str(reflected.server_default.arg), expected[i])
         finally:
-            db.execute("DROP TABLE r_defaults")
+            exec_sql(db, "DROP TABLE r_defaults")
 
     def test_default_reflection_3(self):
         db = testing.db
@@ -513,10 +526,10 @@ class DefaultsTest(fixtures.TestBase, AssertsCompiledSQL):
             val INTEGER NOT NULL DEFAULT 0
             )"""
         try:
-            db.execute(table)
+            exec_sql(db, table)
             m1 = MetaData(db)
             t1 = Table("r_defaults", m1, autoload=True)
-            db.execute("DROP TABLE r_defaults")
+            exec_sql(db, "DROP TABLE r_defaults")
             t1.create()
             m2 = MetaData(db)
             t2 = Table("r_defaults", m2, autoload=True)
@@ -527,7 +540,7 @@ class DefaultsTest(fixtures.TestBase, AssertsCompiledSQL):
                 "NOT NULL)",
             )
         finally:
-            db.execute("DROP TABLE r_defaults")
+            exec_sql(db, "DROP TABLE r_defaults")
 
     @testing.provide_metadata
     def test_boolean_default(self):
@@ -633,14 +646,16 @@ class DialectTest(fixtures.TestBase, AssertsExecutionResults):
         """Tests autoload of tables created with quoted column names."""
 
         metadata = self.metadata
-        testing.db.execute(
+        exec_sql(
+            testing.db,
             """CREATE TABLE "django_content_type" (
             "id" integer NOT NULL PRIMARY KEY,
             "django_stuff" text NULL
         )
-        """
+        """,
         )
-        testing.db.execute(
+        exec_sql(
+            testing.db,
             """
         CREATE TABLE "django_admin_log" (
             "id" integer NOT NULL PRIMARY KEY,
@@ -650,7 +665,7 @@ class DialectTest(fixtures.TestBase, AssertsExecutionResults):
             "object_id" text NULL,
             "change_message" text NOT NULL
         )
-        """
+        """,
         )
         table1 = Table("django_admin_log", metadata, autoload=True)
         table2 = Table("django_content_type", metadata, autoload=True)
@@ -670,16 +685,17 @@ class DialectTest(fixtures.TestBase, AssertsExecutionResults):
         """
 
         metadata = self.metadata
-        testing.db.execute(
+        exec_sql(
+            testing.db,
             r'''CREATE TABLE """a""" (
             """id""" integer NOT NULL PRIMARY KEY
         )
-        '''
+        ''',
         )
 
         # unfortunately, still can't do this; sqlite quadruples
         # up the quotes on the table name here for pragma foreign_key_list
-        # testing.db.execute(r'''
+        # exec_sql(testing.db,r'''
         # CREATE TABLE """b""" (
         #    """id""" integer NOT NULL PRIMARY KEY,
         #    """aid""" integer NULL
@@ -884,7 +900,7 @@ class AttachedDBTest(fixtures.TestBase):
         eq_(insp.get_schema_names(), ["main", "test_schema"])
 
         # implicitly creates a "temp" schema
-        self.conn.execute("select * from sqlite_temp_master")
+        self.conn.exec_driver_sql("select * from sqlite_temp_master")
 
         # we're not including it
         insp = inspect(self.conn)
@@ -1475,8 +1491,8 @@ def full_text_search_missing():
     it is and True otherwise."""
 
     try:
-        testing.db.execute("CREATE VIRTUAL TABLE t using FTS3;")
-        testing.db.execute("DROP TABLE t;")
+        exec_sql(testing.db, "CREATE VIRTUAL TABLE t using FTS3;")
+        exec_sql(testing.db, "DROP TABLE t;")
         return False
     except Exception:
         return True
@@ -1494,17 +1510,19 @@ class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
     def setup_class(cls):
         global metadata, cattable, matchtable
         metadata = MetaData(testing.db)
-        testing.db.execute(
+        exec_sql(
+            testing.db,
             """
         CREATE VIRTUAL TABLE cattable using FTS3 (
             id INTEGER NOT NULL,
             description VARCHAR(50),
             PRIMARY KEY (id)
         )
-        """
+        """,
         )
         cattable = Table("cattable", metadata, autoload=True)
-        testing.db.execute(
+        exec_sql(
+            testing.db,
             """
         CREATE VIRTUAL TABLE matchtable using FTS3 (
             id INTEGER NOT NULL,
@@ -1512,7 +1530,7 @@ class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
             category_id INTEGER NOT NULL,
             PRIMARY KEY (id)
         )
-        """
+        """,
         )
         matchtable = Table("matchtable", metadata, autoload=True)
         metadata.create_all()
@@ -1678,16 +1696,16 @@ class ReflectHeadlessFKsTest(fixtures.TestBase):
     __only_on__ = "sqlite"
 
     def setup(self):
-        testing.db.execute("CREATE TABLE a (id INTEGER PRIMARY KEY)")
+        exec_sql(testing.db, "CREATE TABLE a (id INTEGER PRIMARY KEY)")
         # this syntax actually works on other DBs perhaps we'd want to add
         # tests to test_reflection
-        testing.db.execute(
-            "CREATE TABLE b (id INTEGER PRIMARY KEY REFERENCES a)"
+        exec_sql(
+            testing.db, "CREATE TABLE b (id INTEGER PRIMARY KEY REFERENCES a)"
         )
 
     def teardown(self):
-        testing.db.execute("drop table b")
-        testing.db.execute("drop table a")
+        exec_sql(testing.db, "drop table b")
+        exec_sql(testing.db, "drop table a")
 
     def test_reflect_tables_fk_no_colref(self):
         meta = MetaData()
@@ -1703,17 +1721,21 @@ class KeywordInDatabaseNameTest(fixtures.TestBase):
     @classmethod
     def setup_class(cls):
         with testing.db.begin() as conn:
-            conn.execute('ATTACH %r AS "default"' % conn.engine.url.database)
-            conn.execute('CREATE TABLE "default".a (id INTEGER PRIMARY KEY)')
+            conn.exec_driver_sql(
+                'ATTACH %r AS "default"' % conn.engine.url.database
+            )
+            conn.exec_driver_sql(
+                'CREATE TABLE "default".a (id INTEGER PRIMARY KEY)'
+            )
 
     @classmethod
     def teardown_class(cls):
         with testing.db.begin() as conn:
             try:
-                conn.execute('drop table "default".a')
+                conn.exec_driver_sql('drop table "default".a')
             except Exception:
                 pass
-            conn.execute('DETACH DATABASE "default"')
+            conn.exec_driver_sql('DETACH DATABASE "default"')
 
     def test_reflect(self):
         with testing.db.begin() as conn:
@@ -1729,72 +1751,72 @@ class ConstraintReflectionTest(fixtures.TestBase):
     def setup_class(cls):
         with testing.db.begin() as conn:
 
-            conn.execute("CREATE TABLE a1 (id INTEGER PRIMARY KEY)")
-            conn.execute("CREATE TABLE a2 (id INTEGER PRIMARY KEY)")
-            conn.execute(
+            conn.exec_driver_sql("CREATE TABLE a1 (id INTEGER PRIMARY KEY)")
+            conn.exec_driver_sql("CREATE TABLE a2 (id INTEGER PRIMARY KEY)")
+            conn.exec_driver_sql(
                 "CREATE TABLE b (id INTEGER PRIMARY KEY, "
                 "FOREIGN KEY(id) REFERENCES a1(id),"
                 "FOREIGN KEY(id) REFERENCES a2(id)"
                 ")"
             )
-            conn.execute(
+            conn.exec_driver_sql(
                 "CREATE TABLE c (id INTEGER, "
                 "CONSTRAINT bar PRIMARY KEY(id),"
                 "CONSTRAINT foo1 FOREIGN KEY(id) REFERENCES a1(id),"
                 "CONSTRAINT foo2 FOREIGN KEY(id) REFERENCES a2(id)"
                 ")"
             )
-            conn.execute(
+            conn.exec_driver_sql(
                 # the lower casing + inline is intentional here
                 "CREATE TABLE d (id INTEGER, x INTEGER unique)"
             )
-            conn.execute(
+            conn.exec_driver_sql(
                 # the lower casing + inline is intentional here
                 "CREATE TABLE d1 "
                 '(id INTEGER, "some ( STUPID n,ame" INTEGER unique)'
             )
-            conn.execute(
+            conn.exec_driver_sql(
                 # the lower casing + inline is intentional here
                 'CREATE TABLE d2 ( "some STUPID n,ame" INTEGER unique)'
             )
-            conn.execute(
+            conn.exec_driver_sql(
                 # the lower casing + inline is intentional here
                 'CREATE TABLE d3 ( "some STUPID n,ame" INTEGER NULL unique)'
             )
 
-            conn.execute(
+            conn.exec_driver_sql(
                 # lower casing + inline is intentional
                 "CREATE TABLE e (id INTEGER, x INTEGER references a2(id))"
             )
-            conn.execute(
+            conn.exec_driver_sql(
                 'CREATE TABLE e1 (id INTEGER, "some ( STUPID n,ame" INTEGER '
                 'references a2   ("some ( STUPID n,ame"))'
             )
-            conn.execute(
+            conn.exec_driver_sql(
                 "CREATE TABLE e2 (id INTEGER, "
                 '"some ( STUPID n,ame" INTEGER NOT NULL  '
                 'references a2   ("some ( STUPID n,ame"))'
             )
 
-            conn.execute(
+            conn.exec_driver_sql(
                 "CREATE TABLE f (x INTEGER, CONSTRAINT foo_fx UNIQUE(x))"
             )
-            conn.execute(
+            conn.exec_driver_sql(
                 "CREATE TEMPORARY TABLE g "
                 "(x INTEGER, CONSTRAINT foo_gx UNIQUE(x))"
             )
-            conn.execute(
+            conn.exec_driver_sql(
                 # intentional broken casing
                 "CREATE TABLE h (x INTEGER, COnstraINT foo_hx unIQUE(x))"
             )
-            conn.execute(
+            conn.exec_driver_sql(
                 "CREATE TABLE i (x INTEGER, y INTEGER, PRIMARY KEY(x, y))"
             )
-            conn.execute(
+            conn.exec_driver_sql(
                 "CREATE TABLE j (id INTEGER, q INTEGER, p INTEGER, "
                 "PRIMARY KEY(id), FOreiGN KEY(q,p) REFERENCes  i(x,y))"
             )
-            conn.execute(
+            conn.exec_driver_sql(
                 "CREATE TABLE k (id INTEGER, q INTEGER, p INTEGER, "
                 "PRIMARY KEY(id), "
                 "conSTRAINT my_fk FOreiGN KEY (  q  , p  )   "
@@ -1833,8 +1855,10 @@ class ConstraintReflectionTest(fixtures.TestBase):
             meta.create_all(conn)
 
             # will contain an "autoindex"
-            conn.execute("create table o (foo varchar(20) primary key)")
-            conn.execute(
+            conn.exec_driver_sql(
+                "create table o (foo varchar(20) primary key)"
+            )
+            conn.exec_driver_sql(
                 "CREATE TABLE onud_test (id INTEGER PRIMARY KEY, "
                 "c1 INTEGER, c2 INTEGER, c3 INTEGER, c4 INTEGER, "
                 "CONSTRAINT fk1 FOREIGN KEY (c1) REFERENCES a1(id) "
@@ -1847,30 +1871,30 @@ class ConstraintReflectionTest(fixtures.TestBase):
                 "ON UPDATE NO ACTION)"
             )
 
-            conn.execute(
+            conn.exec_driver_sql(
                 "CREATE TABLE cp ("
                 "q INTEGER check (q > 1 AND q < 6),\n"
                 "CONSTRAINT cq CHECK (q == 1 OR (q > 2 AND q < 5))\n"
                 ")"
             )
 
-            conn.execute(
+            conn.exec_driver_sql(
                 "CREATE TABLE implicit_referred (pk integer primary key)"
             )
             # single col foreign key with no referred column given,
             # must assume primary key of referred table
-            conn.execute(
+            conn.exec_driver_sql(
                 "CREATE TABLE implicit_referrer "
                 "(id integer REFERENCES implicit_referred)"
             )
 
-            conn.execute(
+            conn.exec_driver_sql(
                 "CREATE TABLE implicit_referred_comp "
                 "(pk1 integer, pk2 integer, primary key (pk1, pk2))"
             )
             # composite foreign key with no referred columns given,
             # must assume primary key of referred table
-            conn.execute(
+            conn.exec_driver_sql(
                 "CREATE TABLE implicit_referrer_comp "
                 "(id1 integer, id2 integer, foreign key(id1, id2) "
                 "REFERENCES implicit_referred_comp)"
@@ -1878,7 +1902,7 @@ class ConstraintReflectionTest(fixtures.TestBase):
 
             # worst case - FK that refers to nonexistent table so we cant
             # get pks.  requires FK pragma is turned off
-            conn.execute(
+            conn.exec_driver_sql(
                 "CREATE TABLE implicit_referrer_comp_fake "
                 "(id1 integer, id2 integer, foreign key(id1, id2) "
                 "REFERENCES fake_table)"
@@ -1912,7 +1936,7 @@ class ConstraintReflectionTest(fixtures.TestBase):
                 "a2",
             ]:
                 try:
-                    conn.execute("drop table %s" % name)
+                    conn.exec_driver_sql("drop table %s" % name)
                 except Exception:
                     pass
 
@@ -2199,7 +2223,7 @@ class ConstraintReflectionTest(fixtures.TestBase):
 
     def test_foreign_key_options_unnamed_inline(self):
         with testing.db.connect() as conn:
-            conn.execute(
+            conn.exec_driver_sql(
                 "create table foo (id integer, "
                 "foreign key (id) references bar (id) on update cascade)"
             )
@@ -2365,7 +2389,7 @@ class SavepointTest(fixtures.TablesTest):
         @event.listens_for(engine, "begin")
         def do_begin(conn):
             # emit our own BEGIN
-            conn.execute("BEGIN")
+            conn.exec_driver_sql("BEGIN")
 
         return engine
 
@@ -2536,7 +2560,7 @@ class TypeReflectionTest(fixtures.TestBase):
         conn = testing.db.connect()
         for from_, to_ in self._fixture_as_string(fixture):
             inspector = inspect(conn)
-            conn.execute("CREATE TABLE foo (data %s)" % from_)
+            conn.exec_driver_sql("CREATE TABLE foo (data %s)" % from_)
             try:
                 if warnings:
 
@@ -2559,7 +2583,7 @@ class TypeReflectionTest(fixtures.TestBase):
                             getattr(to_, attr, None),
                         )
             finally:
-                conn.execute("DROP TABLE foo")
+                conn.exec_driver_sql("DROP TABLE foo")
 
     def test_lookup_direct_lookup(self):
         self._test_lookup_direct(self._fixed_lookup_fixture())