From f34f0f1b26cf5296e10f102b79a825eb8899de86 Mon Sep 17 00:00:00 2001 From: khanjan2708 Date: Fri, 20 Feb 2026 20:35:58 +0530 Subject: [PATCH 1/3] Fix duplicate CHECK constraints for Boolean/Enum in batch mode - Suppress automatic type-level events during batch add/alter column - Inherit naming convention in temporary batch metadata - Automatically detect naming convention from migration context Fixes: #1768 --- alembic/operations/base.py | 3 +++ alembic/operations/batch.py | 14 ++++++++++++-- tests/test_batch.py | 34 ++++++++++++++++++++++++++++++++++ 3 files changed, 49 insertions(+), 2 deletions(-) diff --git a/alembic/operations/base.py b/alembic/operations/base.py index b9e6107f..336d09e0 100644 --- a/alembic/operations/base.py +++ b/alembic/operations/base.py @@ -390,6 +390,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, diff --git a/alembic/operations/batch.py b/alembic/operations/batch.py index 9b48be59..f5213ca8 100644 --- a/alembic/operations/batch.py +++ b/alembic/operations/batch.py @@ -322,7 +322,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: @@ -537,6 +537,11 @@ def alter_column( existing.type = type_ + if isinstance(existing.type, SchemaEventTarget): + existing.type._create_events = ( # type:ignore[attr-defined] + existing.type.create_constraint # type:ignore[attr-defined] # noqa + ) = False + # we *dont* however set events for the new type, because # alter_column is invoked from # Operations.implementation_for(alter_column) which already @@ -618,7 +623,12 @@ 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) + if isinstance(new_column.type, SchemaEventTarget): + new_column.type._create_events = ( # type:ignore[attr-defined] + new_column.type.create_constraint # type:ignore[attr-defined] # noqa + ) = False + self.columns[new_column.name] = new_column self.column_transfers[column.name] = {} def drop_column( diff --git a/tests/test_batch.py b/tests/test_batch.py index fa582829..5de6904d 100644 --- a/tests/test_batch.py +++ b/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) From 32587424fc27d3fb601364ea72e40cb4ce66b052 Mon Sep 17 00:00:00 2001 From: khanjan2708 Date: Mon, 23 Feb 2026 14:08:49 +0530 Subject: [PATCH 2/3] Fix PEP 484 type errors and Windows-specific mypy attribute errors --- alembic/operations/base.py | 3 ++- alembic/util/messaging.py | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/alembic/operations/base.py b/alembic/operations/base.py index 336d09e0..1eb3f138 100644 --- a/alembic/operations/base.py +++ b/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. diff --git a/alembic/util/messaging.py b/alembic/util/messaging.py index 4c08f16e..9805e4e5 100644 --- a/alembic/util/messaging.py +++ b/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 From 51c3037f6ead043813b48218a5c3cad0eb8ec799 Mon Sep 17 00:00:00 2001 From: khanjan2708 Date: Mon, 23 Feb 2026 14:18:22 +0530 Subject: [PATCH 3/3] Refactor: move schema event disabling to sqla_compat helper as suggested --- alembic/operations/batch.py | 16 ++++------------ alembic/util/sqla_compat.py | 7 +++++++ 2 files changed, 11 insertions(+), 12 deletions(-) diff --git a/alembic/operations/batch.py b/alembic/operations/batch.py index f5213ca8..8dcdffa1 100644 --- a/alembic/operations/batch.py +++ b/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 @@ -526,10 +527,7 @@ 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_ @@ -537,10 +535,7 @@ def alter_column( existing.type = type_ - 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) # we *dont* however set events for the new type, because # alter_column is invoked from @@ -624,10 +619,7 @@ def add_column( # we copy the column because operations.add_column() # gives us a Column that is part of a Table already. new_column = _copy(column, schema=self.table.schema) - if isinstance(new_column.type, SchemaEventTarget): - new_column.type._create_events = ( # type:ignore[attr-defined] - new_column.type.create_constraint # type:ignore[attr-defined] # noqa - ) = False + _disable_schema_events(new_column.type) self.columns[new_column.name] = new_column self.column_transfers[column.name] = {} diff --git a/alembic/util/sqla_compat.py b/alembic/util/sqla_compat.py index ff2f2c93..8f15900a 100644 --- a/alembic/util/sqla_compat.py +++ b/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