Skip to content

Commit

Permalink
check all directives in batch block until recreate selected
Browse files Browse the repository at this point in the history
Fixed regression in batch mode due to 🎫`883` where the "auto" mode
of batch would fail to accommodate any additional migration directives
beyond encountering an ``add_column()`` directive, due to a mis-application
of the conditional logic that was added as part of this change, leading to
"recreate" mode not being used in cases where it is required for SQLite
such as for unique constraints.

Change-Id: I6315569caff5f3b33d152974ebecc8b18d9cc523
Fixes: #896
  • Loading branch information
zzzeek committed Aug 30, 2021
1 parent ff8fd16 commit 2793416
Show file tree
Hide file tree
Showing 3 changed files with 49 additions and 7 deletions.
1 change: 0 additions & 1 deletion alembic/ddl/sqlite.py
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,6 @@ def requires_recreate_in_batch(
and col.server_default.persisted
):
return True
return False
elif op[0] not in ("create_index", "drop_index"):
return True
else:
Expand Down
10 changes: 10 additions & 0 deletions docs/build/unreleased/896.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
.. change::
:tags: bug, regression, batch
:tickets: 896

Fixed regression in batch mode due to :ticket:`883` where the "auto" mode
of batch would fail to accommodate any additional migration directives
beyond encountering an ``add_column()`` directive, due to a mis-application
of the conditional logic that was added as part of this change, leading to
"recreate" mode not being used in cases where it is required for SQLite
such as for unique constraints.
45 changes: 39 additions & 6 deletions tests/test_batch.py
Original file line number Diff line number Diff line change
Expand Up @@ -1661,19 +1661,21 @@ def test_drop_pk_col_readd_col_also_pk_const(self):
pk_const = inspect(self.conn).get_pk_constraint("foo")
eq_(pk_const["constrained_columns"], ["id"])

def test_add_pk_constraint(self):
@testing.combinations(("always",), ("auto",), argnames="recreate")
def test_add_pk_constraint(self, recreate):
self._no_pk_fixture()
with self.op.batch_alter_table("nopk", recreate="always") as batch_op:
with self.op.batch_alter_table("nopk", recreate=recreate) as batch_op:
batch_op.create_primary_key("newpk", ["a", "b"])

pk_const = inspect(self.conn).get_pk_constraint("nopk")
with config.requirements.reflects_pk_names.fail_if():
eq_(pk_const["name"], "newpk")
eq_(pk_const["constrained_columns"], ["a", "b"])

@testing.combinations(("always",), ("auto",), argnames="recreate")
@config.requirements.check_constraint_reflection
def test_add_ck_constraint(self):
with self.op.batch_alter_table("foo", recreate="always") as batch_op:
def test_add_ck_constraint(self, recreate):
with self.op.batch_alter_table("foo", recreate=recreate) as batch_op:
batch_op.create_check_constraint("newck", text("x > 0"))

ck_consts = inspect(self.conn).get_check_constraints("foo")
Expand All @@ -1682,12 +1684,13 @@ def test_add_ck_constraint(self):
)
eq_(ck_consts, [{"sqltext": "x > 0", "name": "newck"}])

@testing.combinations(("always",), ("auto",), argnames="recreate")
@config.requirements.check_constraint_reflection
def test_drop_ck_constraint(self):
def test_drop_ck_constraint(self, recreate):
self._ck_constraint_fixture()

with self.op.batch_alter_table(
"ck_table", recreate="always"
"ck_table", recreate=recreate
) as batch_op:
batch_op.drop_constraint("ck", "check")

Expand Down Expand Up @@ -1736,6 +1739,36 @@ def _assert_table_comment(self, tname, comment):
tcomment = insp.get_table_comment(tname)
eq_(tcomment, {"text": comment})

@testing.combinations(("always",), ("auto",), argnames="recreate")
def test_add_uq(self, recreate):
with self.op.batch_alter_table("foo", recreate=recreate) as batch_op:
batch_op.create_unique_constraint("newuk", ["x"])

uq_consts = inspect(self.conn).get_unique_constraints("foo")
eq_(
[
{"name": uc["name"], "column_names": uc["column_names"]}
for uc in uq_consts
],
[{"name": "newuk", "column_names": ["x"]}],
)

@testing.combinations(("always",), ("auto",), argnames="recreate")
def test_add_uq_plus_col(self, recreate):
with self.op.batch_alter_table("foo", recreate=recreate) as batch_op:
batch_op.add_column(Column("y", Integer))
batch_op.create_unique_constraint("newuk", ["x", "y"])

uq_consts = inspect(self.conn).get_unique_constraints("foo")

eq_(
[
{"name": uc["name"], "column_names": uc["column_names"]}
for uc in uq_consts
],
[{"name": "newuk", "column_names": ["x", "y"]}],
)

@config.requirements.comments
def test_add_table_comment(self):
with self.op.batch_alter_table("foo") as batch_op:
Expand Down

0 comments on commit 2793416

Please sign in to comment.