Merge "Deprecate plain string in execute and introduce `exec_driver_sql`"

19 files changed, 130 insertions, 107 deletions

diff --git a/lib/sqlalchemy/dialects/firebird/base.py b/lib/sqlalchemy/dialects/firebird/base.py
index 51bda30a2..a43413780 100644
--- a/lib/sqlalchemy/dialects/firebird/base.py
+++ b/lib/sqlalchemy/dialects/firebird/base.py
@@ -41,7 +41,7 @@ hang until other transactions are released.  SQLAlchemy does its best
 to release transactions as quickly as possible.  The most common cause
 of hanging transactions is a non-fully consumed result set, i.e.::
 
-    result = engine.execute("select * from table")
+    result = engine.execute(text("select * from table"))
     row = result.fetchone()
     return
 
@@ -679,7 +679,9 @@ class FBDialect(default.DefaultDialect):
                       FROM rdb$relations
                       WHERE rdb$relation_name=?)
         """
-        c = connection.execute(tblqry, [self.denormalize_name(table_name)])
+        c = connection.exec_driver_sql(
+            tblqry, [self.denormalize_name(table_name)]
+        )
         return c.first() is not None
 
     def has_sequence(self, connection, sequence_name, schema=None):
@@ -691,7 +693,9 @@ class FBDialect(default.DefaultDialect):
                       FROM rdb$generators
                       WHERE rdb$generator_name=?)
         """
-        c = connection.execute(genqry, [self.denormalize_name(sequence_name)])
+        c = connection.exec_driver_sql(
+            genqry, [self.denormalize_name(sequence_name)]
+        )
         return c.first() is not None
 
     @reflection.cache
@@ -714,7 +718,10 @@ class FBDialect(default.DefaultDialect):
         # FROM rdb$relation_fields
         # WHERE rdb$system_flag=0 AND rdb$view_context IS NULL
 
-        return [self.normalize_name(row[0]) for row in connection.execute(s)]
+        return [
+            self.normalize_name(row[0])
+            for row in connection.exec_driver_sql(s)
+        ]
 
     @reflection.cache
     def get_view_names(self, connection, schema=None, **kw):
@@ -725,7 +732,10 @@ class FBDialect(default.DefaultDialect):
         where rdb$view_blr is not null
         and (rdb$system_flag is null or rdb$system_flag = 0);
         """
-        return [self.normalize_name(row[0]) for row in connection.execute(s)]
+        return [
+            self.normalize_name(row[0])
+            for row in connection.exec_driver_sql(s)
+        ]
 
     @reflection.cache
     def get_view_definition(self, connection, view_name, schema=None, **kw):
@@ -734,7 +744,9 @@ class FBDialect(default.DefaultDialect):
         FROM rdb$relations
         WHERE rdb$relation_name=?
         """
-        rp = connection.execute(qry, [self.denormalize_name(view_name)])
+        rp = connection.exec_driver_sql(
+            qry, [self.denormalize_name(view_name)]
+        )
         row = rp.first()
         if row:
             return row["view_source"]
@@ -752,7 +764,7 @@ class FBDialect(default.DefaultDialect):
         """
         tablename = self.denormalize_name(table_name)
         # get primary key fields
-        c = connection.execute(keyqry, ["PRIMARY KEY", tablename])
+        c = connection.exec_driver_sql(keyqry, ["PRIMARY KEY", tablename])
         pkfields = [self.normalize_name(r["fname"]) for r in c.fetchall()]
         return {"constrained_columns": pkfields, "name": None}
 
@@ -780,7 +792,7 @@ class FBDialect(default.DefaultDialect):
            FROM rdb$dependencies trigdep2
            WHERE trigdep2.rdb$dependent_name = trigdep.rdb$dependent_name) = 2
         """
-        genr = connection.execute(genqry, [tablename, colname]).first()
+        genr = connection.exec_driver_sql(genqry, [tablename, colname]).first()
         if genr is not None:
             return dict(name=self.normalize_name(genr["fgenerator"]))
 
@@ -814,7 +826,7 @@ class FBDialect(default.DefaultDialect):
 
         tablename = self.denormalize_name(table_name)
         # get all of the fields for this table
-        c = connection.execute(tblqry, [tablename])
+        c = connection.exec_driver_sql(tblqry, [tablename])
         cols = []
         while True:
             row = c.fetchone()
@@ -905,7 +917,7 @@ class FBDialect(default.DefaultDialect):
         """
         tablename = self.denormalize_name(table_name)
 
-        c = connection.execute(fkqry, ["FOREIGN KEY", tablename])
+        c = connection.exec_driver_sql(fkqry, ["FOREIGN KEY", tablename])
         fks = util.defaultdict(
             lambda: {
                 "name": None,
@@ -944,7 +956,9 @@ class FBDialect(default.DefaultDialect):
           AND rdb$relation_constraints.rdb$constraint_type IS NULL
         ORDER BY index_name, ic.rdb$field_position
         """
-        c = connection.execute(qry, [self.denormalize_name(table_name)])
+        c = connection.exec_driver_sql(
+            qry, [self.denormalize_name(table_name)]
+        )
 
         indexes = util.defaultdict(dict)
         for row in c:
diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 94db296bf..924c0f457 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -2265,9 +2265,9 @@ def _db_plus_owner(fn):
 
 def _switch_db(dbname, connection, fn, *arg, **kw):
     if dbname:
-        current_db = connection.scalar("select db_name()")
+        current_db = connection.exec_driver_sql("select db_name()").scalar()
         if current_db != dbname:
-            connection.execute(
+            connection.exec_driver_sql(
                 "use %s"
                 % connection.dialect.identifier_preparer.quote_schema(dbname)
             )
@@ -2275,7 +2275,7 @@ def _switch_db(dbname, connection, fn, *arg, **kw):
         return fn(*arg, **kw)
     finally:
         if dbname and current_db != dbname:
-            connection.execute(
+            connection.exec_driver_sql(
                 "use %s"
                 % connection.dialect.identifier_preparer.quote_schema(
                     current_db
@@ -2387,7 +2387,7 @@ class MSDialect(default.DefaultDialect):
 
     def do_savepoint(self, connection, name):
         # give the DBAPI a push
-        connection.execute("IF @@TRANCOUNT = 0 BEGIN TRANSACTION")
+        connection.exec_driver_sql("IF @@TRANCOUNT = 0 BEGIN TRANSACTION")
         super(MSDialect, self).do_savepoint(connection, name)
 
     def do_release_savepoint(self, connection, name):
@@ -2751,7 +2751,7 @@ class MSDialect(default.DefaultDialect):
         for col in cols:
             colmap[col["name"]] = col
         # We also run an sp_columns to check for identity columns:
-        cursor = connection.execute(
+        cursor = connection.exec_driver_sql(
             "sp_columns @table_name = '%s', "
             "@table_owner = '%s'" % (tablename, owner)
         )
@@ -2773,7 +2773,7 @@ class MSDialect(default.DefaultDialect):
 
         if ic is not None and self.server_version_info >= MS_2005_VERSION:
             table_fullname = "%s.%s" % (owner, tablename)
-            cursor = connection.execute(
+            cursor = connection.exec_driver_sql(
                 "select ident_seed('%s'), ident_incr('%s')"
                 % (table_fullname, table_fullname)
             )
diff --git a/lib/sqlalchemy/dialects/mssql/provision.py b/lib/sqlalchemy/dialects/mssql/provision.py
index a10043db1..84b9e4194 100644
--- a/lib/sqlalchemy/dialects/mssql/provision.py
+++ b/lib/sqlalchemy/dialects/mssql/provision.py
@@ -15,16 +15,16 @@ def _mssql_update_db_opts(db_url, db_opts):
 @create_db.for_db("mssql")
 def _mssql_create_db(cfg, eng, ident):
     with eng.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
-        conn.execute("create database %s" % ident)
-        conn.execute(
+        conn.exec_driver_sql("create database %s" % ident)
+        conn.exec_driver_sql(
             "ALTER DATABASE %s SET ALLOW_SNAPSHOT_ISOLATION ON" % ident
         )
-        conn.execute(
+        conn.exec_driver_sql(
             "ALTER DATABASE %s SET READ_COMMITTED_SNAPSHOT ON" % ident
         )
-        conn.execute("use %s" % ident)
-        conn.execute("create schema test_schema")
-        conn.execute("create schema test_schema_2")
+        conn.exec_driver_sql("use %s" % ident)
+        conn.exec_driver_sql("create schema test_schema")
+        conn.exec_driver_sql("create schema test_schema_2")
 
 
 @drop_db.for_db("mssql")
@@ -37,13 +37,13 @@ def _mssql_drop_ignore(conn, ident):
     try:
         # typically when this happens, we can't KILL the session anyway,
         # so let the cleanup process drop the DBs
-        # for row in conn.execute(
+        # for row in conn.exec_driver_sql(
         #     "select session_id from sys.dm_exec_sessions "
         #        "where database_id=db_id('%s')" % ident):
         #    log.info("killing SQL server sesssion %s", row['session_id'])
-        #    conn.execute("kill %s" % row['session_id'])
+        #    conn.exec_driver_sql("kill %s" % row['session_id'])
 
-        conn.execute("drop database %s" % ident)
+        conn.exec_driver_sql("drop database %s" % ident)
         log.info("Reaped db: %s", ident)
         return True
     except exc.DatabaseError as err:
@@ -59,7 +59,7 @@ def _reap_mssql_dbs(url, idents):
 
         log.info("identifiers in file: %s", ", ".join(idents))
 
-        to_reap = conn.execute(
+        to_reap = conn.exec_driver_sql(
             "select d.name from sys.databases as d where name "
             "like 'TEST_%' and not exists (select session_id "
             "from sys.dm_exec_sessions "
diff --git a/lib/sqlalchemy/dialects/mssql/pymssql.py b/lib/sqlalchemy/dialects/mssql/pymssql.py
index b8b3bf67d..2d355d964 100644
--- a/lib/sqlalchemy/dialects/mssql/pymssql.py
+++ b/lib/sqlalchemy/dialects/mssql/pymssql.py
@@ -73,7 +73,7 @@ class MSDialect_pymssql(MSDialect):
         return module
 
     def _get_server_version_info(self, connection):
-        vers = connection.scalar("select @@version")
+        vers = connection.exec_driver_sql("select @@version").scalar()
         m = re.match(r"Microsoft .*? - (\d+).(\d+).(\d+).(\d+)", vers)
         if m:
             return tuple(int(x) for x in m.group(1, 2, 3, 4))
diff --git a/lib/sqlalchemy/dialects/mssql/pyodbc.py b/lib/sqlalchemy/dialects/mssql/pyodbc.py
index 4ba3a0dfa..ffaf66fdd 100644
--- a/lib/sqlalchemy/dialects/mssql/pyodbc.py
+++ b/lib/sqlalchemy/dialects/mssql/pyodbc.py
@@ -353,9 +353,9 @@ class MSDialect_pyodbc(PyODBCConnector, MSDialect):
         try:
             # "Version of the instance of SQL Server, in the form
             # of 'major.minor.build.revision'"
-            raw = connection.scalar(
+            raw = connection.exec_driver_sql(
                 "SELECT CAST(SERVERPROPERTY('ProductVersion') AS VARCHAR)"
-            )
+            ).scalar()
         except exc.DBAPIError:
             # SQL Server docs indicate this function isn't present prior to
             # 2008.  Before we had the VARCHAR cast above, pyodbc would also
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index a97719532..e193c1daa 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -2371,7 +2371,7 @@ class MySQLDialect(default.DefaultDialect):
         connection.execute(sql.text("XA COMMIT :xid"), xid=xid)
 
     def do_recover_twophase(self, connection):
-        resultset = connection.execute("XA RECOVER")
+        resultset = connection.exec_driver_sql("XA RECOVER")
         return [row["data"][0 : row["gtrid_length"]] for row in resultset]
 
     def is_disconnect(self, e, connection, cursor):
@@ -2424,7 +2424,7 @@ class MySQLDialect(default.DefaultDialect):
         raise NotImplementedError()
 
     def _get_default_schema_name(self, connection):
-        return connection.execute("SELECT DATABASE()").scalar()
+        return connection.exec_driver_sql("SELECT DATABASE()").scalar()
 
     def has_table(self, connection, table_name, schema=None):
         # SHOW TABLE STATUS LIKE and SHOW TABLES LIKE do not function properly
@@ -2449,7 +2449,7 @@ class MySQLDialect(default.DefaultDialect):
             try:
                 rs = connection.execution_options(
                     skip_user_error_events=True
-                ).execute(st)
+                ).exec_driver_sql(st)
                 have = rs.fetchone() is not None
                 rs.close()
                 return have
@@ -2552,7 +2552,7 @@ class MySQLDialect(default.DefaultDialect):
 
     @reflection.cache
     def get_schema_names(self, connection, **kw):
-        rp = connection.execute("SHOW schemas")
+        rp = connection.exec_driver_sql("SHOW schemas")
         return [r[0] for r in rp]
 
     @reflection.cache
@@ -2565,7 +2565,7 @@ class MySQLDialect(default.DefaultDialect):
 
         charset = self._connection_charset
         if self.server_version_info < (5, 0, 2):
-            rp = connection.execute(
+            rp = connection.exec_driver_sql(
                 "SHOW TABLES FROM %s"
                 % self.identifier_preparer.quote_identifier(current_schema)
             )
@@ -2573,7 +2573,7 @@ class MySQLDialect(default.DefaultDialect):
                 row[0] for row in self._compat_fetchall(rp, charset=charset)
             ]
         else:
-            rp = connection.execute(
+            rp = connection.exec_driver_sql(
                 "SHOW FULL TABLES FROM %s"
                 % self.identifier_preparer.quote_identifier(current_schema)
             )
@@ -2593,7 +2593,7 @@ class MySQLDialect(default.DefaultDialect):
         if self.server_version_info < (5, 0, 2):
             return self.get_table_names(connection, schema)
         charset = self._connection_charset
-        rp = connection.execute(
+        rp = connection.exec_driver_sql(
             "SHOW FULL TABLES FROM %s"
             % self.identifier_preparer.quote_identifier(schema)
         )
@@ -2905,7 +2905,9 @@ class MySQLDialect(default.DefaultDialect):
 
         charset = self._connection_charset
         row = self._compat_first(
-            connection.execute("SHOW VARIABLES LIKE 'lower_case_table_names'"),
+            connection.execute(
+                sql.text("SHOW VARIABLES LIKE 'lower_case_table_names'")
+            ),
             charset=charset,
         )
         if not row:
@@ -2933,14 +2935,14 @@ class MySQLDialect(default.DefaultDialect):
             pass
         else:
             charset = self._connection_charset
-            rs = connection.execute("SHOW COLLATION")
+            rs = connection.exec_driver_sql("SHOW COLLATION")
             for row in self._compat_fetchall(rs, charset):
                 collations[row[0]] = row[1]
         return collations
 
     def _detect_sql_mode(self, connection):
         row = self._compat_first(
-            connection.execute("SHOW VARIABLES LIKE 'sql_mode'"),
+            connection.exec_driver_sql("SHOW VARIABLES LIKE 'sql_mode'"),
             charset=self._connection_charset,
         )
 
@@ -2981,7 +2983,7 @@ class MySQLDialect(default.DefaultDialect):
         try:
             rp = connection.execution_options(
                 skip_user_error_events=True
-            ).execute(st)
+            ).exec_driver_sql(st)
         except exc.DBAPIError as e:
             if self._extract_error_code(e.orig) == 1146:
                 util.raise_(exc.NoSuchTableError(full_name), replace_context=e)
@@ -3004,7 +3006,7 @@ class MySQLDialect(default.DefaultDialect):
             try:
                 rp = connection.execution_options(
                     skip_user_error_events=True
-                ).execute(st)
+                ).exec_driver_sql(st)
             except exc.DBAPIError as e:
                 code = self._extract_error_code(e.orig)
                 if code == 1146:
diff --git a/lib/sqlalchemy/dialects/mysql/mysqldb.py b/lib/sqlalchemy/dialects/mysql/mysqldb.py
index 56aa991db..03c1779c3 100644
--- a/lib/sqlalchemy/dialects/mysql/mysqldb.py
+++ b/lib/sqlalchemy/dialects/mysql/mysqldb.py
@@ -154,15 +154,14 @@ class MySQLDialect_mysqldb(MySQLDialect):
         # https://github.com/farcepest/MySQLdb1/commit/cd44524fef63bd3fcb71947392326e9742d520e8
         # specific issue w/ the utf8mb4_bin collation and unicode returns
 
-        has_utf8mb4_bin = self.server_version_info > (
-            5,
-        ) and connection.scalar(
+        collation = connection.exec_driver_sql(
             "show collation where %s = 'utf8mb4' and %s = 'utf8mb4_bin'"
             % (
                 self.identifier_preparer.quote("Charset"),
                 self.identifier_preparer.quote("Collation"),
             )
-        )
+        ).scalar()
+        has_utf8mb4_bin = self.server_version_info > (5,) and collation
         if has_utf8mb4_bin:
             additional_tests = [
                 sql.collate(
diff --git a/lib/sqlalchemy/dialects/mysql/oursql.py b/lib/sqlalchemy/dialects/mysql/oursql.py
index 32357a24b..d7334711b 100644
--- a/lib/sqlalchemy/dialects/mysql/oursql.py
+++ b/lib/sqlalchemy/dialects/mysql/oursql.py
@@ -91,7 +91,7 @@ class MySQLDialect_oursql(MySQLDialect):
                 xid.encode(charset)
             ).decode(charset)
         arg = "'%s'" % arg
-        connection.execution_options(_oursql_plain_query=True).execute(
+        connection.execution_options(_oursql_plain_query=True).exec_driver_sql(
             query % arg
         )
 
diff --git a/lib/sqlalchemy/dialects/mysql/provision.py b/lib/sqlalchemy/dialects/mysql/provision.py
index 843b35caa..bf126464d 100644
--- a/lib/sqlalchemy/dialects/mysql/provision.py
+++ b/lib/sqlalchemy/dialects/mysql/provision.py
@@ -12,11 +12,13 @@ def _mysql_create_db(cfg, eng, ident):
         except Exception:
             pass
 
-        conn.execute("CREATE DATABASE %s CHARACTER SET utf8mb4" % ident)
-        conn.execute(
+        conn.exec_driver_sql(
+            "CREATE DATABASE %s CHARACTER SET utf8mb4" % ident
+        )
+        conn.exec_driver_sql(
             "CREATE DATABASE %s_test_schema CHARACTER SET utf8mb4" % ident
         )
-        conn.execute(
+        conn.exec_driver_sql(
             "CREATE DATABASE %s_test_schema_2 CHARACTER SET utf8mb4" % ident
         )
 
@@ -30,9 +32,9 @@ def _mysql_configure_follower(config, ident):
 @drop_db.for_db("mysql")
 def _mysql_drop_db(cfg, eng, ident):
     with eng.connect() as conn:
-        conn.execute("DROP DATABASE %s_test_schema" % ident)
-        conn.execute("DROP DATABASE %s_test_schema_2" % ident)
-        conn.execute("DROP DATABASE %s" % ident)
+        conn.exec_driver_sql("DROP DATABASE %s_test_schema" % ident)
+        conn.exec_driver_sql("DROP DATABASE %s_test_schema_2" % ident)
+        conn.exec_driver_sql("DROP DATABASE %s" % ident)
 
 
 @temp_table_keyword_args.for_db("mysql")
diff --git a/lib/sqlalchemy/dialects/mysql/pyodbc.py b/lib/sqlalchemy/dialects/mysql/pyodbc.py
index eb62e6425..5a696562e 100644
--- a/lib/sqlalchemy/dialects/mysql/pyodbc.py
+++ b/lib/sqlalchemy/dialects/mysql/pyodbc.py
@@ -87,7 +87,9 @@ class MySQLDialect_pyodbc(PyODBCConnector, MySQLDialect):
         #
         # If it's decided that issuing that sort of SQL leaves you SOL, then
         # this can prefer the driver value.
-        rs = connection.execute("SHOW VARIABLES LIKE 'character_set%%'")
+        rs = connection.exec_driver_sql(
+            "SHOW VARIABLES LIKE 'character_set%%'"
+        )
         opts = {row[0]: row[1] for row in self._compat_fetchall(rs)}
         for key in ("character_set_connection", "character_set"):
             if opts.get(key, None):
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 356c37562..d12453924 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -1340,7 +1340,7 @@ class OracleDialect(default.DefaultDialect):
         if self.server_version_info < (12, 2):
             return self.server_version_info
         try:
-            compat = connection.execute(
+            compat = connection.exec_driver_sql(
                 "SELECT value FROM v$parameter WHERE name = 'compatible'"
             ).scalar()
         except exc.DBAPIError:
@@ -1424,7 +1424,7 @@ class OracleDialect(default.DefaultDialect):
 
     def _get_default_schema_name(self, connection):
         return self.normalize_name(
-            connection.execute("SELECT USER FROM DUAL").scalar()
+            connection.exec_driver_sql("SELECT USER FROM DUAL").scalar()
         )
 
     def _resolve_synonym(
@@ -1535,7 +1535,7 @@ class OracleDialect(default.DefaultDialect):
     @reflection.cache
     def get_schema_names(self, connection, **kw):
         s = "SELECT username FROM all_users ORDER BY username"
-        cursor = connection.execute(s)
+        cursor = connection.exec_driver_sql(s)
         return [self.normalize_name(row[0]) for row in cursor]
 
     @reflection.cache
diff --git a/lib/sqlalchemy/dialects/oracle/cx_oracle.py b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
index 69423992f..d5177daa6 100644
--- a/lib/sqlalchemy/dialects/oracle/cx_oracle.py
+++ b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
@@ -870,10 +870,10 @@ class OracleDialect_cx_oracle(OracleDialect):
         # NLS_TERRITORY or formatting behavior of the DB, we opt
         # to just look at it
 
-        self._decimal_char = connection.scalar(
+        self._decimal_char = connection.exec_driver_sql(
             "select value from nls_session_parameters "
             "where parameter = 'NLS_NUMERIC_CHARACTERS'"
-        )[0]
+        ).scalar()[0]
         if self._decimal_char != ".":
             _detect_decimal = self._detect_decimal
             _to_decimal = self._to_decimal
diff --git a/lib/sqlalchemy/dialects/oracle/provision.py b/lib/sqlalchemy/dialects/oracle/provision.py
index 7901eb4e8..9de14bff0 100644
--- a/lib/sqlalchemy/dialects/oracle/provision.py
+++ b/lib/sqlalchemy/dialects/oracle/provision.py
@@ -17,13 +17,13 @@ def _oracle_create_db(cfg, eng, ident):
     # similar, so that the default tablespace is not "system"; reflection will
     # fail otherwise
     with eng.connect() as conn:
-        conn.execute("create user %s identified by xe" % ident)
-        conn.execute("create user %s_ts1 identified by xe" % ident)
-        conn.execute("create user %s_ts2 identified by xe" % ident)
-        conn.execute("grant dba to %s" % (ident,))
-        conn.execute("grant unlimited tablespace to %s" % ident)
-        conn.execute("grant unlimited tablespace to %s_ts1" % ident)
-        conn.execute("grant unlimited tablespace to %s_ts2" % ident)
+        conn.exec_driver_sql("create user %s identified by xe" % ident)
+        conn.exec_driver_sql("create user %s_ts1 identified by xe" % ident)
+        conn.exec_driver_sql("create user %s_ts2 identified by xe" % ident)
+        conn.exec_driver_sql("grant dba to %s" % (ident,))
+        conn.exec_driver_sql("grant unlimited tablespace to %s" % ident)
+        conn.exec_driver_sql("grant unlimited tablespace to %s_ts1" % ident)
+        conn.exec_driver_sql("grant unlimited tablespace to %s_ts2" % ident)
 
 
 @configure_follower.for_db("oracle")
@@ -34,7 +34,7 @@ def _oracle_configure_follower(config, ident):
 
 def _ora_drop_ignore(conn, dbname):
     try:
-        conn.execute("drop user %s cascade" % dbname)
+        conn.exec_driver_sql("drop user %s cascade" % dbname)
         log.info("Reaped db: %s", dbname)
         return True
     except exc.DatabaseError as err:
@@ -68,7 +68,7 @@ def _reap_oracle_dbs(url, idents):
 
         log.info("identifiers in file: %s", ", ".join(idents))
 
-        to_reap = conn.execute(
+        to_reap = conn.exec_driver_sql(
             "select u.username from all_users u where username "
             "like 'TEST_%' and not exists (select username "
             "from v$session where username=u.username)"
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 6089b2b8a..cb41a8f65 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -195,10 +195,10 @@ in order to determine the remote schema name.  That is, if our ``search_path``
 were set to include ``test_schema``, and we invoked a table
 reflection process as follows::
 
-    >>> from sqlalchemy import Table, MetaData, create_engine
+    >>> from sqlalchemy import Table, MetaData, create_engine, text
     >>> engine = create_engine("postgresql://scott:tiger@localhost/test")
     >>> with engine.connect() as conn:
-    ...     conn.execute("SET search_path TO test_schema, public")
+    ...     conn.execute(text("SET search_path TO test_schema, public"))
     ...     meta = MetaData()
     ...     referring = Table('referring', meta,
     ...                       autoload=True, autoload_with=conn)
@@ -218,7 +218,7 @@ dialect-specific argument to both :class:`.Table` as well as
 :meth:`.MetaData.reflect`::
 
     >>> with engine.connect() as conn:
-    ...     conn.execute("SET search_path TO test_schema, public")
+    ...     conn.execute(text("SET search_path TO test_schema, public"))
     ...     meta = MetaData()
     ...     referring = Table('referring', meta, autoload=True,
     ...                       autoload_with=conn,
@@ -2464,9 +2464,11 @@ class PGDialect(default.DefaultDialect):
         # http://www.postgresql.org/docs/9.3/static/release-9-2.html#AEN116689
         self.supports_smallserial = self.server_version_info >= (9, 2)
 
+        std_string = connection.exec_driver_sql(
+            "show standard_conforming_strings"
+        ).scalar()
         self._backslash_escapes = (
-            self.server_version_info < (8, 2)
-            or connection.scalar("show standard_conforming_strings") == "off"
+            self.server_version_info < (8, 2) or std_string == "off"
         )
 
         self._supports_create_index_concurrently = (
@@ -2523,7 +2525,7 @@ class PGDialect(default.DefaultDialect):
         self.do_begin(connection.connection)
 
     def do_prepare_twophase(self, connection, xid):
-        connection.execute("PREPARE TRANSACTION '%s'" % xid)
+        connection.exec_driver_sql("PREPARE TRANSACTION '%s'" % xid)
 
     def do_rollback_twophase(
         self, connection, xid, is_prepared=True, recover=False
@@ -2534,9 +2536,9 @@ class PGDialect(default.DefaultDialect):
                 # context when committing recoverable transactions
                 # Must find out a way how to make the dbapi not
                 # open a transaction.
-                connection.execute("ROLLBACK")
-            connection.execute("ROLLBACK PREPARED '%s'" % xid)
-            connection.execute("BEGIN")
+                connection.exec_driver_sql("ROLLBACK")
+            connection.exec_driver_sql("ROLLBACK PREPARED '%s'" % xid)
+            connection.exec_driver_sql("BEGIN")
             self.do_rollback(connection.connection)
         else:
             self.do_rollback(connection.connection)
@@ -2546,9 +2548,9 @@ class PGDialect(default.DefaultDialect):
     ):
         if is_prepared:
             if recover:
-                connection.execute("ROLLBACK")
-            connection.execute("COMMIT PREPARED '%s'" % xid)
-            connection.execute("BEGIN")
+                connection.exec_driver_sql("ROLLBACK")
+            connection.exec_driver_sql("COMMIT PREPARED '%s'" % xid)
+            connection.exec_driver_sql("BEGIN")
             self.do_rollback(connection.connection)
         else:
             self.do_commit(connection.connection)
@@ -2560,7 +2562,7 @@ class PGDialect(default.DefaultDialect):
         return [row[0] for row in resultset]
 
     def _get_default_schema_name(self, connection):
-        return connection.scalar("select current_schema()")
+        return connection.exec_driver_sql("select current_schema()").scalar()
 
     def has_schema(self, connection, schema):
         query = (
@@ -2689,7 +2691,7 @@ class PGDialect(default.DefaultDialect):
         return bool(cursor.scalar())
 
     def _get_server_version_info(self, connection):
-        v = connection.execute("select version()").scalar()
+        v = connection.exec_driver_sql("select version()").scalar()
         m = re.match(
             r".*(?:PostgreSQL|EnterpriseDB) "
             r"(\d+)\.?(\d+)?(?:\.(\d+))?(?:\.\d+)?(?:devel|beta)?",
diff --git a/lib/sqlalchemy/dialects/postgresql/provision.py b/lib/sqlalchemy/dialects/postgresql/provision.py
index c8f83d2be..6c6dc4be6 100644
--- a/lib/sqlalchemy/dialects/postgresql/provision.py
+++ b/lib/sqlalchemy/dialects/postgresql/provision.py
@@ -18,12 +18,14 @@ def _pg_create_db(cfg, eng, ident):
         except Exception:
             pass
         if not template_db:
-            template_db = conn.scalar("select current_database()")
+            template_db = conn.exec_driver_sql(
+                "select current_database()"
+            ).scalar()
 
         attempt = 0
         while True:
             try:
-                conn.execute(
+                conn.exec_driver_sql(
                     "CREATE DATABASE %s TEMPLATE %s" % (ident, template_db)
                 )
             except exc.OperationalError as err:
@@ -56,7 +58,7 @@ def _pg_drop_db(cfg, eng, ident):
             ),
             dname=ident,
         )
-        conn.execute("DROP DATABASE %s" % ident)
+        conn.exec_driver_sql("DROP DATABASE %s" % ident)
 
 
 @temp_table_keyword_args.for_db("postgresql")
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index b1a83bf92..a63ce0033 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -518,14 +518,14 @@ to filter these out::
     eng = create_engine("sqlite://")
     conn = eng.connect()
 
-    conn.execute("create table x (a integer, b integer)")
-    conn.execute("insert into x (a, b) values (1, 1)")
-    conn.execute("insert into x (a, b) values (2, 2)")
+    conn.exec_driver_sql("create table x (a integer, b integer)")
+    conn.exec_driver_sql("insert into x (a, b) values (1, 1)")
+    conn.exec_driver_sql("insert into x (a, b) values (2, 2)")
 
-    result = conn.execute("select x.a, x.b from x")
+    result = conn.exec_driver_sql("select x.a, x.b from x")
     assert result.keys() == ["a", "b"]
 
-    result = conn.execute('''
+    result = conn.exec_driver_sql('''
         select x.a, x.b from x where a=1
         union
         select x.a, x.b from x where a=2
@@ -553,7 +553,7 @@ contain dots, and the functionality of :meth:`.ResultProxy.keys` and
 the ``sqlite_raw_colnames`` execution option may be provided, either on a
 per-:class:`.Connection` basis::
 
-    result = conn.execution_options(sqlite_raw_colnames=True).execute('''
+    result = conn.execution_options(sqlite_raw_colnames=True).exec_driver_sql('''
         select x.a, x.b from x where a=1
         union
         select x.a, x.b from x where a=2
@@ -1588,7 +1588,7 @@ class SQLiteDialect(default.DefaultDialect):
     @reflection.cache
     def get_schema_names(self, connection, **kw):
         s = "PRAGMA database_list"
-        dl = connection.execute(s)
+        dl = connection.exec_driver_sql(s)
 
         return [db[1] for db in dl if db[1] != "temp"]
 
@@ -1602,7 +1602,7 @@ class SQLiteDialect(default.DefaultDialect):
         s = ("SELECT name FROM %s " "WHERE type='table' ORDER BY name") % (
             master,
         )
-        rs = connection.execute(s)
+        rs = connection.exec_driver_sql(s)
         return [row[0] for row in rs]
 
     @reflection.cache
@@ -1611,7 +1611,7 @@ class SQLiteDialect(default.DefaultDialect):
             "SELECT name FROM sqlite_temp_master "
             "WHERE type='table' ORDER BY name "
         )
-        rs = connection.execute(s)
+        rs = connection.exec_driver_sql(s)
 
         return [row[0] for row in rs]
 
@@ -1621,7 +1621,7 @@ class SQLiteDialect(default.DefaultDialect):
             "SELECT name FROM sqlite_temp_master "
             "WHERE type='view' ORDER BY name "
         )
-        rs = connection.execute(s)
+        rs = connection.exec_driver_sql(s)
 
         return [row[0] for row in rs]
 
@@ -1641,7 +1641,7 @@ class SQLiteDialect(default.DefaultDialect):
         s = ("SELECT name FROM %s " "WHERE type='view' ORDER BY name") % (
             master,
         )
-        rs = connection.execute(s)
+        rs = connection.exec_driver_sql(s)
 
         return [row[0] for row in rs]
 
@@ -1654,7 +1654,7 @@ class SQLiteDialect(default.DefaultDialect):
                 master,
                 view_name,
             )
-            rs = connection.execute(s)
+            rs = connection.exec_driver_sql(s)
         else:
             try:
                 s = (
@@ -1664,13 +1664,13 @@ class SQLiteDialect(default.DefaultDialect):
                     "WHERE name = '%s' "
                     "AND type='view'"
                 ) % view_name
-                rs = connection.execute(s)
+                rs = connection.exec_driver_sql(s)
             except exc.DBAPIError:
                 s = (
                     "SELECT sql FROM sqlite_master WHERE name = '%s' "
                     "AND type='view'"
                 ) % view_name
-                rs = connection.execute(s)
+                rs = connection.exec_driver_sql(s)
 
         result = rs.fetchall()
         if result:
@@ -2070,7 +2070,7 @@ class SQLiteDialect(default.DefaultDialect):
                 "AND type = 'table'"
                 % {"schema": schema_expr, "table": table_name}
             )
-            rs = connection.execute(s)
+            rs = connection.exec_driver_sql(s)
         except exc.DBAPIError:
             s = (
                 "SELECT sql FROM %(schema)ssqlite_master "
@@ -2078,7 +2078,7 @@ class SQLiteDialect(default.DefaultDialect):
                 "AND type = 'table'"
                 % {"schema": schema_expr, "table": table_name}
             )
-            rs = connection.execute(s)
+            rs = connection.exec_driver_sql(s)
         return rs.scalar()
 
     def _get_table_pragma(self, connection, pragma, table_name, schema=None):
@@ -2095,7 +2095,7 @@ class SQLiteDialect(default.DefaultDialect):
         qtable = quote(table_name)
         for statement in statements:
             statement = "%s%s(%s)" % (statement, pragma, qtable)
-            cursor = connection.execute(statement)
+            cursor = connection.exec_driver_sql(statement)
             if not cursor._soft_closed:
                 # work around SQLite issue whereby cursor.description
                 # is blank when PRAGMA returns no rows:
diff --git a/lib/sqlalchemy/dialects/sqlite/pysqlcipher.py b/lib/sqlalchemy/dialects/sqlite/pysqlcipher.py
index f8236dea9..a1243f271 100644
--- a/lib/sqlalchemy/dialects/sqlite/pysqlcipher.py
+++ b/lib/sqlalchemy/dialects/sqlite/pysqlcipher.py
@@ -112,10 +112,10 @@ class SQLiteDialect_pysqlcipher(SQLiteDialect_pysqlite):
         conn = super(SQLiteDialect_pysqlcipher, self).connect(
             *cargs, **cparams
         )
-        conn.execute('pragma key="%s"' % passphrase)
+        conn.exec_driver_sql('pragma key="%s"' % passphrase)
         for prag, value in pragmas.items():
             if value is not None:
-                conn.execute('pragma %s="%s"' % (prag, value))
+                conn.exec_driver_sql('pragma %s="%s"' % (prag, value))
 
         return conn
 
diff --git a/lib/sqlalchemy/dialects/sqlite/pysqlite.py b/lib/sqlalchemy/dialects/sqlite/pysqlite.py
index 4485631ce..b8a42a506 100644
--- a/lib/sqlalchemy/dialects/sqlite/pysqlite.py
+++ b/lib/sqlalchemy/dialects/sqlite/pysqlite.py
@@ -322,7 +322,7 @@ ourselves. This is achieved using two event listeners::
     @event.listens_for(engine, "begin")
     def do_begin(conn):
         # emit our own BEGIN
-        conn.execute("BEGIN")
+        conn.exec_driver_sql("BEGIN")
 
 Above, we intercept a new pysqlite connection and disable any transactional
 integration.   Then, at the point at which SQLAlchemy knows that transaction
@@ -335,7 +335,7 @@ by adding the desired locking mode to our ``"BEGIN"``::
 
     @event.listens_for(engine, "begin")
     def do_begin(conn):
-        conn.execute("BEGIN EXCLUSIVE")
+        conn.exec_driver_sql("BEGIN EXCLUSIVE")
 
 .. seealso::
 
diff --git a/lib/sqlalchemy/dialects/sybase/pysybase.py b/lib/sqlalchemy/dialects/sybase/pysybase.py
index ab8e6eac0..a36cd74ca 100644
--- a/lib/sqlalchemy/dialects/sybase/pysybase.py
+++ b/lib/sqlalchemy/dialects/sybase/pysybase.py
@@ -82,7 +82,7 @@ class SybaseDialect_pysybase(SybaseDialect):
             cursor.execute(statement, param)
 
     def _get_server_version_info(self, connection):
-        vers = connection.scalar("select @@version_number")
+        vers = connection.exec_driver_sql("select @@version_number").scalar()
         # i.e. 15500, 15000, 12500 == (15, 5, 0, 0), (15, 0, 0, 0),
         # (12, 5, 0, 0)
         return (vers / 1000, vers % 1000 / 100, vers % 100 / 10, vers % 10)