Fix duplicate CHECK constraints for Boolean/Enum in batch mode

alembic/operations/base.py

@@ -38,6 +38,7 @@
     from typing import Literal
 
     from sqlalchemy import Table
+    from sqlalchemy import Constraint
     from sqlalchemy.engine import Connection
     from sqlalchemy.sql import Executable
     from sqlalchemy.sql.expression import ColumnElement
@@ -252,7 +253,7 @@ def batch_alter_table(
         table_kwargs: Mapping[str, Any] = util.immutabledict(),
         reflect_args: Tuple[Any, ...] = (),
         reflect_kwargs: Mapping[str, Any] = util.immutabledict(),
-        naming_convention: Optional[Dict[str, str]] = None,
+        naming_convention: Optional[Mapping[Any, Any]] = None,
     ) -> Iterator[BatchOperations]:
         """Invoke a series of per-table migrations in batch.
 
@@ -390,6 +391,9 @@ def batch_alter_table(
             :ref:`batch_migrations`
 
         """
+        if naming_convention is None:
+            naming_convention = self.schema_obj.metadata().naming_convention
+
         impl = batch.BatchOperationsImpl(
             self,
             table_name,

alembic/operations/batch.py

@@ -29,6 +29,7 @@
 from ..util.sqla_compat import _columns_for_constraint
 from ..util.sqla_compat import _copy
 from ..util.sqla_compat import _copy_expression
+from ..util.sqla_compat import _disable_schema_events
 from ..util.sqla_compat import _ensure_scope_for_ddl
 from ..util.sqla_compat import _fk_is_self_referential
 from ..util.sqla_compat import _idx_table_bound_expressions
@@ -322,7 +323,7 @@ def _adjust_self_columns_for_partial_reordering(self) -> None:
     def _transfer_elements_to_new_table(self) -> None:
         assert self.new_table is None, "Can only create new table once"
 
-        m = MetaData()
+        m = MetaData(naming_convention=self.table.metadata.naming_convention)
         schema = self.table.schema
 
         if self.partial_reordering or self.add_col_ordering:
@@ -526,17 +527,16 @@ def alter_column(
             # we also ignore the drop_constraint that will come here from
             # Operations.implementation_for(alter_column)
 
-            if isinstance(existing.type, SchemaEventTarget):
-                existing.type._create_events = (  # type:ignore[attr-defined]
-                    existing.type.create_constraint  # type:ignore[attr-defined] # noqa
-                ) = False
+            _disable_schema_events(existing.type)
 
             self.impl.cast_for_batch_migrate(
                 existing, existing_transfer, type_
             )
 
             existing.type = type_
 
+            _disable_schema_events(existing.type)
+
             # we *dont* however set events for the new type, because
             # alter_column is invoked from
             # Operations.implementation_for(alter_column) which already
@@ -618,7 +618,9 @@ def add_column(
         )
         # we copy the column because operations.add_column()
         # gives us a Column that is part of a Table already.
-        self.columns[column.name] = _copy(column, schema=self.table.schema)
+        new_column = _copy(column, schema=self.table.schema)
+        _disable_schema_events(new_column.type)
+        self.columns[new_column.name] = new_column
         self.column_transfers[column.name] = {}
 
     def drop_column(

alembic/util/messaging.py

@@ -24,7 +24,7 @@
     import termios
     import struct
 
-    ioctl = fcntl.ioctl(0, termios.TIOCGWINSZ, struct.pack("HHHH", 0, 0, 0, 0))
+    ioctl = fcntl.ioctl(0, termios.TIOCGWINSZ, struct.pack("HHHH", 0, 0, 0, 0))  # type: ignore[attr-defined]
     _h, TERMWIDTH, _hp, _wp = struct.unpack("HHHH", ioctl)
     if TERMWIDTH <= 0:  # can occur if running in emacs pseudo-tty
         TERMWIDTH = None

alembic/util/sqla_compat.py

@@ -508,3 +508,10 @@ def _inherit_schema_deprecated() -> bool:
     # at some point in 2.1 inherit_schema was replaced with a property
     # so that's preset at the class level, while before it wasn't.
     return sqla_2_1 and hasattr(sqltypes.Enum, "inherit_schema")
+
+
+def _disable_schema_events(type_: Any) -> None:
+    if hasattr(type_, "_create_events"):
+        type_._create_events = False
+    if hasattr(type_, "create_constraint"):
+        type_.create_constraint = False

tests/test_batch.py

@@ -413,6 +413,40 @@ def test_rename_col_boolean(self):
             1,
         )
 
+    def test_add_column_boolean_convention(self):
+        m = MetaData(
+            naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
+        )
+        t = Table(
+            "tname",
+            m,
+            Column("id", Integer, primary_key=True),
+        )
+        impl = ApplyBatchImpl(self.impl, t, (), {}, False)
+        impl.add_column(
+            "tname",
+            Column("flag", Boolean(create_constraint=True, name="ck1")),
+        )
+        ck = CheckConstraint("flag IN (0, 1)", name="ck1")
+        t.append_constraint(ck)
+        impl.add_constraint(ck)
+        new_table = self._assert_impl(
+            impl,
+            ddl_contains="CONSTRAINT ck_tname_ck1 CHECK (flag IN (0, 1))",
+            colnames=["id", "flag"],
+            dialect="sqlite",
+        )
+        eq_(
+            len(
+                [
+                    const
+                    for const in new_table.constraints
+                    if isinstance(const, CheckConstraint)
+                ]
+            ),
+            1,
+        )
+
     def test_change_type_schematype_to_non(self):
         impl = self._boolean_fixture()
         impl.alter_column("tname", "flag", type_=Integer)