From f24a644e15216980917ca9160fe1dcc5f3c040aa Mon Sep 17 00:00:00 2001
From: Mike Bayer <mike_mp@zzzcomputing.com>
Date: Wed, 10 Jan 2024 10:13:16 -0500
Subject: [PATCH] dont pass empty sequences to connection.execute()

Fixed internal issue where Alembic would call ``connection.execute()``
sending an empty tuple to indicate "no params".  In SQLAlchemy 2.1 this
case will be deprecated as "empty sequence" is ambiguous as to its intent.

Fixes: #1394
Change-Id: If3105866a13f4e3ffdcd513de3f970257ea48089
---
 alembic/ddl/impl.py            | 24 ++++++++++++++----------
 alembic/testing/fixtures.py    |  6 ++++++
 docs/build/unreleased/1394.rst |  8 ++++++++
 tests/test_impl.py             | 27 +++++++++++++++++++++++++++
 4 files changed, 55 insertions(+), 10 deletions(-)
 create mode 100644 docs/build/unreleased/1394.rst

diff --git a/alembic/ddl/impl.py b/alembic/ddl/impl.py
index 2e4f1ae9..bf202a48 100644
--- a/alembic/ddl/impl.py
+++ b/alembic/ddl/impl.py
@@ -168,16 +168,15 @@ class DefaultImpl(metaclass=ImplMeta):
     def _exec(
         self,
         construct: Union[Executable, str],
-        execution_options: Optional[dict[str, Any]] = None,
-        multiparams: Sequence[dict] = (),
-        params: Dict[str, Any] = util.immutabledict(),
+        execution_options: Optional[Mapping[str, Any]] = None,
+        multiparams: Optional[Sequence[Mapping[str, Any]]] = None,
+        params: Mapping[str, Any] = util.immutabledict(),
     ) -> Optional[CursorResult]:
         if isinstance(construct, str):
             construct = text(construct)
         if self.as_sql:
-            if multiparams or params:
-                # TODO: coverage
-                raise Exception("Execution arguments not allowed with as_sql")
+            if multiparams is not None or params:
+                raise TypeError("SQL parameters not allowed with as_sql")
 
             compile_kw: dict[str, Any]
             if self.literal_binds and not isinstance(
@@ -200,11 +199,16 @@ class DefaultImpl(metaclass=ImplMeta):
             assert conn is not None
             if execution_options:
                 conn = conn.execution_options(**execution_options)
-            if params:
-                assert isinstance(multiparams, tuple)
-                multiparams += (params,)
 
-            return conn.execute(construct, multiparams)
+            if params and multiparams is not None:
+                raise TypeError(
+                    "Can't send params and multiparams at the same time"
+                )
+
+            if multiparams:
+                return conn.execute(construct, multiparams)
+            else:
+                return conn.execute(construct, params)
 
     def execute(
         self,
diff --git a/alembic/testing/fixtures.py b/alembic/testing/fixtures.py
index 4b83a745..b6cea632 100644
--- a/alembic/testing/fixtures.py
+++ b/alembic/testing/fixtures.py
@@ -49,6 +49,12 @@ class TestBase(SQLAlchemyTestBase):
             connection, opts=dict(transaction_per_migration=True)
         )
 
+    @testing.fixture
+    def as_sql_migration_context(self, connection):
+        return MigrationContext.configure(
+            connection, opts=dict(transaction_per_migration=True, as_sql=True)
+        )
+
     @testing.fixture
     def connection(self):
         with config.db.connect() as conn:
diff --git a/docs/build/unreleased/1394.rst b/docs/build/unreleased/1394.rst
new file mode 100644
index 00000000..7e597709
--- /dev/null
+++ b/docs/build/unreleased/1394.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, execution
+    :tickets: 1394
+
+    Fixed internal issue where Alembic would call ``connection.execute()``
+    sending an empty tuple to indicate "no params".  In SQLAlchemy 2.1 this
+    case will be deprecated as "empty sequence" is ambiguous as to its intent.
+
diff --git a/tests/test_impl.py b/tests/test_impl.py
index 8a73b87d..ffd5f61c 100644
--- a/tests/test_impl.py
+++ b/tests/test_impl.py
@@ -23,6 +23,11 @@ class ImplTest(TablesTest):
         with migration_context.begin_transaction(_per_migration=True):
             yield migration_context.impl
 
+    @testing.fixture
+    def as_sql_impl(self, as_sql_migration_context):
+        with as_sql_migration_context.begin_transaction(_per_migration=True):
+            yield as_sql_migration_context.impl
+
     def test_execute_params(self, impl):
         result = impl._exec(text("select :my_param"), params={"my_param": 5})
         eq_(result.scalar(), 5)
@@ -40,6 +45,28 @@ class ImplTest(TablesTest):
             [(1, 2), (2, 3), (5, 7)],
         )
 
+    def test_dont_send_both(self, impl):
+        with testing.expect_raises_message(
+            TypeError, "Can't send params and multiparams at the same time"
+        ):
+            impl._exec(
+                text("select :my_param"),
+                params={"my_param": 5},
+                multiparams=[],
+            )
+
+    def test_no_params_w_as_sql(self, as_sql_impl):
+        with testing.expect_raises_message(
+            TypeError, "SQL parameters not allowed with as_sql"
+        ):
+            as_sql_impl._exec(text("select :my_param"), params={"my_param": 5})
+
+    def test_no_multiparams_w_as_sql(self, as_sql_impl):
+        with testing.expect_raises_message(
+            TypeError, "SQL parameters not allowed with as_sql"
+        ):
+            as_sql_impl._exec(text("select :my_param"), multiparams=[])
+
 
 class FutureImplTest(FutureEngineMixin, ImplTest):
     pass
-- 
2.47.2