diff options
| -rw-r--r-- | alembic/operations/batch.py | 23 | ||||
| -rw-r--r-- | alembic/util/sqla_compat.py | 8 | ||||
| -rw-r--r-- | docs/build/unreleased/982.rst | 7 | ||||
| -rw-r--r-- | tests/test_batch.py | 24 | ||||
| -rw-r--r-- | tox.ini | 1 |
5 files changed, 55 insertions, 8 deletions
diff --git a/alembic/operations/batch.py b/alembic/operations/batch.py index ef182fd..b163d79 100644 --- a/alembic/operations/batch.py +++ b/alembic/operations/batch.py @@ -28,6 +28,7 @@ from ..util.sqla_compat import _fk_is_self_referential from ..util.sqla_compat import _insert_inline from ..util.sqla_compat import _is_type_bound from ..util.sqla_compat import _remove_column_from_collection +from ..util.sqla_compat import _resolve_for_variant from ..util.sqla_compat import _select if TYPE_CHECKING: @@ -462,16 +463,22 @@ class ApplyBatchImpl: existing.name = name existing_transfer["name"] = name - # pop named constraints for Boolean/Enum for rename - if ( - "existing_type" in kw - and isinstance(kw["existing_type"], SchemaEventTarget) - and kw["existing_type"].name # type:ignore[attr-defined] - ): - self.named_constraints.pop( - kw["existing_type"].name, None # type:ignore[attr-defined] + existing_type = kw.get("existing_type", None) + if existing_type: + resolved_existing_type = _resolve_for_variant( + kw["existing_type"], self.impl.dialect ) + # pop named constraints for Boolean/Enum for rename + if ( + isinstance(resolved_existing_type, SchemaEventTarget) + and resolved_existing_type.name # type:ignore[attr-defined] # noqa E501 + ): + self.named_constraints.pop( + resolved_existing_type.name, + None, # type:ignore[attr-defined] + ) + if type_ is not None: type_ = sqltypes.to_instance(type_) # old type is being discarded so turn off eventing diff --git a/alembic/util/sqla_compat.py b/alembic/util/sqla_compat.py index 0f2762c..0f8c12f 100644 --- a/alembic/util/sqla_compat.py +++ b/alembic/util/sqla_compat.py @@ -250,6 +250,14 @@ def _reflect_table( return inspector.reflecttable(table, None) +def _resolve_for_variant(type_, dialect): + if _type_has_variants(type_): + base_type, mapping = _get_variant_mapping(type_) + return mapping.get(dialect.name, base_type) + else: + return type_ + + if hasattr(sqltypes.TypeEngine, "_variant_mapping"): def _type_has_variants(type_): diff --git a/docs/build/unreleased/982.rst b/docs/build/unreleased/982.rst new file mode 100644 index 0000000..715cb1f --- /dev/null +++ b/docs/build/unreleased/982.rst @@ -0,0 +1,7 @@ +.. change:: + :tags: bug, batch, regression + :tickets: 982 + + Fixed regression where usage of a ``with_variant()`` datatype in + conjunction with the ``existing_type`` option of ``op.alter_column()`` + under batch mode would lead to an internal exception. diff --git a/tests/test_batch.py b/tests/test_batch.py index 700056a..f0bbd75 100644 --- a/tests/test_batch.py +++ b/tests/test_batch.py @@ -1138,6 +1138,30 @@ class CopyFromTest(TestBase): "ALTER TABLE _alembic_tmp_foo RENAME TO foo", ) + def test_change_name_from_existing_variant_type(self): + """test #982""" + context = self._fixture() + self.table.append_column( + Column("y", Text().with_variant(Text(10000), "mysql")) + ) + + with self.op.batch_alter_table( + "foo", copy_from=self.table + ) as batch_op: + batch_op.alter_column( + column_name="y", + new_column_name="q", + existing_type=Text().with_variant(Text(10000), "mysql"), + ) + context.assert_( + "CREATE TABLE _alembic_tmp_foo (id INTEGER NOT NULL, " + "data VARCHAR(50), x INTEGER, q TEXT, PRIMARY KEY (id))", + "INSERT INTO _alembic_tmp_foo (id, data, x, q) " + "SELECT foo.id, foo.data, foo.x, foo.y FROM foo", + "DROP TABLE foo", + "ALTER TABLE _alembic_tmp_foo RENAME TO foo", + ) + def test_change_type_to_schematype(self): context = self._fixture() self.table.append_column(Column("y", Integer)) @@ -10,6 +10,7 @@ cov_args=--cov=alembic --cov-report term --cov-report xml deps=pytest>4.6 pytest-xdist + sqla13: pytest<7 sqla13: {[tox]SQLA_REPO}@rel_1_3#egg=sqlalchemy sqla14: {[tox]SQLA_REPO}@rel_1_4#egg=sqlalchemy sqlamain: {[tox]SQLA_REPO}#egg=sqlalchemy |