diff --git a/alembic/operations/base.py b/alembic/operations/base.py index b9e6107f..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. @@ -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, diff --git a/alembic/operations/batch.py b/alembic/operations/batch.py index 9b48be59..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 @@ -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,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,6 +535,8 @@ def alter_column( 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( 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 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 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)