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

7 files changed, 94 insertions, 67 deletions

diff --git a/test/sql/test_defaults.py b/test/sql/test_defaults.py
index 957fa890a..7d33933f8 100644
--- a/test/sql/test_defaults.py
+++ b/test/sql/test_defaults.py
@@ -1654,9 +1654,11 @@ class SequenceAsServerDefaultTest(
 
     def test_default_textual_w_default(self):
         with testing.db.connect() as conn:
-            conn.execute("insert into t_seq_test (data) values ('some data')")
+            conn.exec_driver_sql(
+                "insert into t_seq_test (data) values ('some data')"
+            )
 
-            eq_(conn.scalar("select id from t_seq_test"), 1)
+            eq_(conn.exec_driver_sql("select id from t_seq_test").scalar(), 1)
 
     def test_default_core_w_default(self):
         t_seq_test = self.tables.t_seq_test
@@ -1667,11 +1669,13 @@ class SequenceAsServerDefaultTest(
 
     def test_default_textual_server_only(self):
         with testing.db.connect() as conn:
-            conn.execute(
+            conn.exec_driver_sql(
                 "insert into t_seq_test_2 (data) values ('some data')"
             )
 
-            eq_(conn.scalar("select id from t_seq_test_2"), 1)
+            eq_(
+                conn.exec_driver_sql("select id from t_seq_test_2").scalar(), 1
+            )
 
     def test_default_core_server_only(self):
         t_seq_test = self.tables.t_seq_test_2
diff --git a/test/sql/test_from_linter.py b/test/sql/test_from_linter.py
index bf2f06b57..416f89de3 100644
--- a/test/sql/test_from_linter.py
+++ b/test/sql/test_from_linter.py
@@ -219,7 +219,7 @@ class TestLinter(fixtures.TablesTest):
 
     def test_noop_for_unhandled_objects(self):
         with self.bind.connect() as conn:
-            conn.execute("SELECT 1;").fetchone()
+            conn.exec_driver_sql("SELECT 1;").fetchone()
 
     def test_does_not_modify_query(self):
         with self.bind.connect() as conn:
diff --git a/test/sql/test_quote.py b/test/sql/test_quote.py
index aba6a0204..627626994 100644
--- a/test/sql/test_quote.py
+++ b/test/sql/test_quote.py
@@ -90,19 +90,23 @@ class QuoteExecTest(fixtures.TestBase):
     @testing.provide_metadata
     def test_has_table_case_sensitive(self):
         preparer = testing.db.dialect.identifier_preparer
-        if testing.db.dialect.requires_name_normalize:
-            testing.db.execute("CREATE TABLE TAB1 (id INTEGER)")
-        else:
-            testing.db.execute("CREATE TABLE tab1 (id INTEGER)")
-        testing.db.execute(
-            "CREATE TABLE %s (id INTEGER)" % preparer.quote_identifier("tab2")
-        )
-        testing.db.execute(
-            "CREATE TABLE %s (id INTEGER)" % preparer.quote_identifier("TAB3")
-        )
-        testing.db.execute(
-            "CREATE TABLE %s (id INTEGER)" % preparer.quote_identifier("TAB4")
-        )
+        with testing.db.connect() as conn:
+            if conn.dialect.requires_name_normalize:
+                conn.exec_driver_sql("CREATE TABLE TAB1 (id INTEGER)")
+            else:
+                conn.exec_driver_sql("CREATE TABLE tab1 (id INTEGER)")
+            conn.exec_driver_sql(
+                "CREATE TABLE %s (id INTEGER)"
+                % preparer.quote_identifier("tab2")
+            )
+            conn.exec_driver_sql(
+                "CREATE TABLE %s (id INTEGER)"
+                % preparer.quote_identifier("TAB3")
+            )
+            conn.exec_driver_sql(
+                "CREATE TABLE %s (id INTEGER)"
+                % preparer.quote_identifier("TAB4")
+            )
 
         t1 = Table(
             "tab1", self.metadata, Column("id", Integer, primary_key=True)
diff --git a/test/sql/test_resultset.py b/test/sql/test_resultset.py
index 87886c4fa..f08248440 100644
--- a/test/sql/test_resultset.py
+++ b/test/sql/test_resultset.py
@@ -586,8 +586,8 @@ class ResultProxyTest(fixtures.TablesTest):
             )
             trans.rollback()
 
-    def test_fetchone_til_end(self):
-        result = testing.db.execute("select * from users")
+    def test_fetchone_til_end(self, connection):
+        result = connection.exec_driver_sql("select * from users")
         eq_(result.fetchone(), None)
         eq_(result.fetchone(), None)
         eq_(result.fetchone(), None)
@@ -600,9 +600,10 @@ class ResultProxyTest(fixtures.TablesTest):
 
     def test_connectionless_autoclose_rows_exhausted(self):
         users = self.tables.users
-        users.insert().execute(dict(user_id=1, user_name="john"))
+        with testing.db.connect() as conn:
+            conn.execute(users.insert(), dict(user_id=1, user_name="john"))
 
-        result = testing.db.execute("select * from users")
+        result = testing.db.execute(text("select * from users"))
         connection = result.connection
         assert not connection.closed
         eq_(result.fetchone(), (1, "john"))
@@ -627,7 +628,7 @@ class ResultProxyTest(fixtures.TablesTest):
         assert connection.closed
 
     def test_connectionless_autoclose_no_rows(self):
-        result = testing.db.execute("select * from users")
+        result = testing.db.execute(text("select * from users"))
         connection = result.connection
         assert not connection.closed
         eq_(result.fetchone(), None)
@@ -635,7 +636,7 @@ class ResultProxyTest(fixtures.TablesTest):
 
     @testing.requires.updateable_autoincrement_pks
     def test_connectionless_autoclose_no_metadata(self):
-        result = testing.db.execute("update users set user_id=5")
+        result = testing.db.execute(text("update users set user_id=5"))
         connection = result.connection
         assert connection.closed
         assert_raises_message(
@@ -1071,16 +1072,18 @@ class ResultProxyTest(fixtures.TablesTest):
             [("user_id", 1), ("user_name", "foo")],
         )
 
-    def test_len(self):
+    def test_len(self, connection):
         users = self.tables.users
 
-        users.insert().execute(user_id=1, user_name="foo")
-        r = users.select().execute().first()
+        connection.execute(users.insert(), dict(user_id=1, user_name="foo"))
+        r = connection.execute(users.select()).first()
         eq_(len(r), 2)
 
-        r = testing.db.execute("select user_name, user_id from users").first()
+        r = connection.exec_driver_sql(
+            "select user_name, user_id from users"
+        ).first()
         eq_(len(r), 2)
-        r = testing.db.execute("select user_name from users").first()
+        r = connection.exec_driver_sql("select user_name from users").first()
         eq_(len(r), 1)
 
     def test_sorting_in_python(self):
@@ -1109,12 +1112,15 @@ class ResultProxyTest(fixtures.TablesTest):
         eq_([x.lower() for x in r._fields], ["user_id", "user_name"])
         eq_(list(r._mapping.values()), [1, "foo"])
 
-    def test_column_order_with_text_query(self):
+    def test_column_order_with_text_query(self, connection):
         # should return values in query order
         users = self.tables.users
 
-        users.insert().execute(user_id=1, user_name="foo")
-        r = testing.db.execute("select user_name, user_id from users").first()
+        connection.execute(users.insert(), dict(user_id=1, user_name="foo"))
+
+        r = connection.exec_driver_sql(
+            "select user_name, user_id from users"
+        ).first()
         eq_(r[0], "foo")
         eq_(r[1], 1)
         eq_([x.lower() for x in r._fields], ["user_name", "user_id"])
@@ -1271,8 +1277,10 @@ class ResultProxyTest(fixtures.TablesTest):
         eq_(row[1:0:-1], ("Uno",))
 
     @testing.only_on("sqlite")
-    def test_row_getitem_indexes_raw(self):
-        row = testing.db.execute("select 'One' as key, 'Uno' as value").first()
+    def test_row_getitem_indexes_raw(self, connection):
+        row = connection.exec_driver_sql(
+            "select 'One' as key, 'Uno' as value"
+        ).first()
         eq_(row._mapping["key"], "One")
         eq_(row._mapping["value"], "Uno")
         eq_(row[0], "One")
@@ -1304,7 +1312,7 @@ class ResultProxyTest(fixtures.TablesTest):
         assert s.getvalue().strip() == "1,Test"
 
     @testing.requires.selectone
-    def test_empty_accessors(self):
+    def test_empty_accessors(self, connection):
         statements = [
             (
                 "select 1",
@@ -1339,7 +1347,10 @@ class ResultProxyTest(fixtures.TablesTest):
         ]
 
         for stmt, meths, msg in statements:
-            r = testing.db.execute(stmt)
+            if isinstance(stmt, str):
+                r = connection.exec_driver_sql(stmt)
+            else:
+                r = connection.execute(stmt)
             try:
                 for meth in meths:
                     assert_raises_message(
diff --git a/test/sql/test_returning.py b/test/sql/test_returning.py
index 4cfb3f0d6..d81ad7186 100644
--- a/test/sql/test_returning.py
+++ b/test/sql/test_returning.py
@@ -171,13 +171,13 @@ class ReturningTest(fixtures.TestBase, AssertsExecutionResults):
         )
 
     @testing.fails_on_everything_except("postgresql", "firebird")
-    def test_literal_returning(self):
+    def test_literal_returning(self, connection):
         if testing.against("postgresql"):
             literal_true = "true"
         else:
             literal_true = "1"
 
-        result4 = testing.db.execute(
+        result4 = connection.exec_driver_sql(
             'insert into tables (id, persons, "full") '
             "values (5, 10, %s) returning persons" % literal_true
         )
diff --git a/test/sql/test_type_expressions.py b/test/sql/test_type_expressions.py
index 8c3e8c5d4..412623c01 100644
--- a/test/sql/test_type_expressions.py
+++ b/test/sql/test_type_expressions.py
@@ -333,8 +333,8 @@ class DerivedTest(_ExprFixture, fixtures.TestBase, AssertsCompiledSQL):
 
 
 class RoundTripTestBase(object):
-    def test_round_trip(self):
-        testing.db.execute(
+    def test_round_trip(self, connection):
+        connection.execute(
             self.tables.test_table.insert(),
             {"x": "X1", "y": "Y1"},
             {"x": "X2", "y": "Y2"},
@@ -343,7 +343,7 @@ class RoundTripTestBase(object):
 
         # test insert coercion alone
         eq_(
-            testing.db.execute(
+            connection.exec_driver_sql(
                 "select * from test_table order by y"
             ).fetchall(),
             [("X1", "y1"), ("X2", "y2"), ("X3", "y3")],
@@ -351,7 +351,7 @@ class RoundTripTestBase(object):
 
         # conversion back to upper
         eq_(
-            testing.db.execute(
+            connection.execute(
                 select([self.tables.test_table]).order_by(
                     self.tables.test_table.c.y
                 )
diff --git a/test/sql/test_types.py b/test/sql/test_types.py
index 356470dd3..9fb79958d 100644
--- a/test/sql/test_types.py
+++ b/test/sql/test_types.py
@@ -1520,7 +1520,7 @@ class EnumTest(AssertsCompiledSQL, fixtures.TablesTest):
             eq_(conn.scalar(select([non_native_enum_table.c.someenum])), None)
 
     @testing.requires.enforces_check_constraints
-    def test_check_constraint(self):
+    def test_check_constraint(self, connection):
         assert_raises(
             (
                 exc.IntegrityError,
@@ -1530,7 +1530,7 @@ class EnumTest(AssertsCompiledSQL, fixtures.TablesTest):
                 # https://github.com/PyMySQL/PyMySQL/issues/607 is resolved
                 exc.InternalError,
             ),
-            testing.db.execute,
+            connection.exec_driver_sql,
             "insert into non_native_enum_table "
             "(id, someenum) values(1, 'four')",
         )
@@ -1563,10 +1563,10 @@ class EnumTest(AssertsCompiledSQL, fixtures.TablesTest):
             self.metadata.create_all(conn)
             assert_raises(
                 (exc.DBAPIError,),
-                conn.execute,
+                conn.exec_driver_sql,
                 "insert into my_table " "(data) values('four')",
             )
-            conn.execute("insert into my_table (data) values ('two')")
+            conn.exec_driver_sql("insert into my_table (data) values ('two')")
 
     @testing.requires.enforces_check_constraints
     @testing.provide_metadata
@@ -1596,19 +1596,21 @@ class EnumTest(AssertsCompiledSQL, fixtures.TablesTest):
             self.metadata.create_all(conn)
             assert_raises(
                 (exc.DBAPIError,),
-                conn.execute,
+                conn.exec_driver_sql,
                 "insert into my_table " "(data) values('two')",
             )
-            conn.execute("insert into my_table (data) values ('four')")
+            conn.exec_driver_sql("insert into my_table (data) values ('four')")
 
     def test_skip_check_constraint(self):
         with testing.db.connect() as conn:
-            conn.execute(
+            conn.exec_driver_sql(
                 "insert into non_native_enum_table "
                 "(id, someotherenum) values(1, 'four')"
             )
             eq_(
-                conn.scalar("select someotherenum from non_native_enum_table"),
+                conn.exec_driver_sql(
+                    "select someotherenum from non_native_enum_table"
+                ).scalar(),
                 "four",
             )
             assert_raises_message(
@@ -2275,11 +2277,14 @@ class ExpressionTest(
     def teardown_class(cls):
         meta.drop_all()
 
-    def test_control(self):
-        assert testing.db.execute("select avalue from test").scalar() == 250
+    def test_control(self, connection):
+        assert (
+            connection.exec_driver_sql("select avalue from test").scalar()
+            == 250
+        )
 
         eq_(
-            test_table.select().execute().fetchall(),
+            connection.execute(test_table.select()).fetchall(),
             [
                 (
                     1,
@@ -2787,35 +2792,35 @@ class NumericRawSQLTest(fixtures.TestBase):
 
     @testing.fails_on("sqlite", "Doesn't provide Decimal results natively")
     @testing.provide_metadata
-    def test_decimal_fp(self):
+    def test_decimal_fp(self, connection):
         metadata = self.metadata
         self._fixture(metadata, Numeric(10, 5), decimal.Decimal("45.5"))
-        val = testing.db.execute("select val from t").scalar()
+        val = connection.exec_driver_sql("select val from t").scalar()
         assert isinstance(val, decimal.Decimal)
         eq_(val, decimal.Decimal("45.5"))
 
     @testing.fails_on("sqlite", "Doesn't provide Decimal results natively")
     @testing.provide_metadata
-    def test_decimal_int(self):
+    def test_decimal_int(self, connection):
         metadata = self.metadata
         self._fixture(metadata, Numeric(10, 5), decimal.Decimal("45"))
-        val = testing.db.execute("select val from t").scalar()
+        val = connection.exec_driver_sql("select val from t").scalar()
         assert isinstance(val, decimal.Decimal)
         eq_(val, decimal.Decimal("45"))
 
     @testing.provide_metadata
-    def test_ints(self):
+    def test_ints(self, connection):
         metadata = self.metadata
         self._fixture(metadata, Integer, 45)
-        val = testing.db.execute("select val from t").scalar()
+        val = connection.exec_driver_sql("select val from t").scalar()
         assert isinstance(val, util.int_types)
         eq_(val, 45)
 
     @testing.provide_metadata
-    def test_float(self):
+    def test_float(self, connection):
         metadata = self.metadata
         self._fixture(metadata, Float, 46.583)
-        val = testing.db.execute("select val from t").scalar()
+        val = connection.exec_driver_sql("select val from t").scalar()
         assert isinstance(val, float)
 
         # some DBAPIs have unusual float handling
@@ -2936,16 +2941,16 @@ class BooleanTest(
     )
     @testing.fails_on("mssql", "FIXME: MS-SQL 2005 doesn't honor CHECK ?!?")
     @testing.skip_if(lambda: testing.db.dialect.supports_native_boolean)
-    def test_constraint(self):
+    def test_constraint(self, connection):
         assert_raises(
             (exc.IntegrityError, exc.ProgrammingError),
-            testing.db.execute,
+            connection.exec_driver_sql,
             "insert into boolean_table (id, value) values(1, 5)",
         )
 
     @testing.skip_if(lambda: testing.db.dialect.supports_native_boolean)
-    def test_unconstrained(self):
-        testing.db.execute(
+    def test_unconstrained(self, connection):
+        connection.exec_driver_sql(
             "insert into boolean_table (id, unconstrained_value)"
             "values (1, 5)"
         )
@@ -2993,13 +2998,16 @@ class BooleanTest(
     def test_nonnative_processor_coerces_integer_to_boolean(self):
         boolean_table = self.tables.boolean_table
         with testing.db.connect() as conn:
-            conn.execute(
+            conn.exec_driver_sql(
                 "insert into boolean_table (id, unconstrained_value) "
                 "values (1, 5)"
             )
 
             eq_(
-                conn.scalar("select unconstrained_value from boolean_table"), 5
+                conn.exec_driver_sql(
+                    "select unconstrained_value from boolean_table"
+                ).scalar(),
+                5,
             )
 
             eq_(