Fixed bug in the check constraint detection implemented in :ticket:`508`
that failed to take into account column bound check constraints,
leading to wrong autogenerate detections.
Fixes: #1842
Change-Id: I43e18fb3b54ffbabc3e59bd1a7270516c7b1713f
inspector = autogen_context.inspector
impl = autogen_context.migration_context.impl
inspector = autogen_context.inspector
impl = autogen_context.migration_context.impl
- metadata_ck_constraints = {
- ck
- for ck in metadata_table.constraints
- if isinstance(ck, sa_schema.CheckConstraint)
- and not sqla_compat._is_type_bound(ck)
- }
+ metadata_ck_constraints = sqla_compat.all_table_check_constraints(
+ metadata_table
+ )
try:
conn_ck_list = _InspectorConv(inspector).get_check_constraints(
try:
conn_ck_list = _InspectorConv(inspector).get_check_constraints(
# 1a. get raw indexes and unique constraints from metadata ...
if metadata_table is not None:
# 1a. get raw indexes and unique constraints from metadata ...
if metadata_table is not None:
+ # unique constraints can't be set on columns, are all table bound.
metadata_unique_constraints = {
uq
for uq in metadata_table.constraints
metadata_unique_constraints = {
uq
for uq in metadata_table.constraints
return PriorityDispatchResult.CONTINUE
inspector = autogen_context.inspector
return PriorityDispatchResult.CONTINUE
inspector = autogen_context.inspector
+ # fk constraints can be set on columns, but they appear also on
+ # the table constraint list
metadata_fks = {
fk
for fk in metadata_table.constraints
metadata_fks = {
fk
for fk in metadata_table.constraints
from sqlalchemy.testing import uses_deprecated
from sqlalchemy.testing.config import combinations
from sqlalchemy.testing.config import fixture
from sqlalchemy.testing import uses_deprecated
from sqlalchemy.testing.config import combinations
from sqlalchemy.testing.config import fixture
-from sqlalchemy.testing.config import requirements as requires
from sqlalchemy.testing.config import Variation
from sqlalchemy.testing.config import variation
from sqlalchemy.testing.config import Variation
from sqlalchemy.testing.config import variation
from collections.abc import Iterable
from collections.abc import Iterator
import contextlib
from collections.abc import Iterable
from collections.abc import Iterator
import contextlib
+from itertools import chain
import re
from typing import Any
from typing import Callable
import re
from typing import Any
from typing import Callable
from sqlalchemy.schema import CheckConstraint
from sqlalchemy.schema import Column
from sqlalchemy.schema import ForeignKeyConstraint
from sqlalchemy.schema import CheckConstraint
from sqlalchemy.schema import Column
from sqlalchemy.schema import ForeignKeyConstraint
+from sqlalchemy.schema import Table
from sqlalchemy.sql import visitors
from sqlalchemy.sql.base import DialectKWArgs
from sqlalchemy.sql.elements import BindParameter
from sqlalchemy.sql import visitors
from sqlalchemy.sql.base import DialectKWArgs
from sqlalchemy.sql.elements import BindParameter
from sqlalchemy import ClauseElement
from sqlalchemy import Identity
from sqlalchemy import Index
from sqlalchemy import ClauseElement
from sqlalchemy import Identity
from sqlalchemy import Index
- from sqlalchemy import Table
from sqlalchemy.engine import Connection
from sqlalchemy.engine import Dialect
from sqlalchemy.engine import Transaction
from sqlalchemy.engine import Connection
from sqlalchemy.engine import Dialect
from sqlalchemy.engine import Transaction
table = constraint.parent
assert table is not None
return table # type: ignore[return-value]
table = constraint.parent
assert table is not None
return table # type: ignore[return-value]
+ elif isinstance(constraint, CheckConstraint):
+ parent = constraint.parent
+ if not isinstance(parent, Table):
+ table = parent.table
+ assert table is not None
+ return table
+ else:
+ return parent
else:
return constraint.table
else:
return constraint.table
# 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")
# 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 all_table_check_constraints(table: Table) -> set[CheckConstraint]:
+ """Returns all check constraint that are not type-bound,
+ including those on columns."""
+ candidates = chain(
+ table.constraints, *(c.constraints for c in table.columns)
+ )
+ return {
+ ck
+ for ck in candidates
+ if isinstance(ck, CheckConstraint) and not _is_type_bound(ck)
+ }
--- /dev/null
+.. change::
+ :tags: bug, autogenerate
+ :tickets: 1842
+
+ Fixed bug in the check constraint detection implemented in :ticket:`508`
+ that failed to take into account column bound check constraints,
+ leading to wrong autogenerate detections.
self._mysql_and_check_constraints_exist,
)
self._mysql_and_check_constraints_exist,
)
+ @property
+ def inline_check_constraint_reflection(self):
+ return exclusions.only_on(
+ [
+ "postgresql",
+ "sqlite",
+ "oracle",
+ ]
+ )
+
def mysql_check_col_name_change(self, config):
# MySQL has check constraints that enforce an reflect, however
# they prevent a column's name from being changed due to a bug in
def mysql_check_col_name_change(self, config):
# MySQL has check constraints that enforce an reflect, however
# they prevent a column's name from being changed due to a bug in
from alembic.testing import config
from alembic.testing import eq_
from alembic.testing import eq_ignore_whitespace
from alembic.testing import config
from alembic.testing import eq_
from alembic.testing import eq_ignore_whitespace
+from alembic.testing import fixture
from alembic.testing import TestBase
from alembic.testing import util
from alembic.testing.env import clear_staging_env
from alembic.testing import TestBase
from alembic.testing import util
from alembic.testing.env import clear_staging_env
+@fixture(params=["table", "column"])
+def col_and_check(request):
+ if (
+ request.param == "column"
+ and not config.requirements.inline_check_constraint_reflection.enabled
+ ):
+ config.skip_test("does not support column bound check constraints")
+
+ def make(name, type_, check):
+ if request.param == "table":
+ return [Column(name, type_), check]
+ else:
+ return [Column(name, type_, check)]
+
+ return make
+
+
class AutogenCheckConstraintTest(AutogenFixtureTest, TestBase):
__backend__ = True
__requires__ = ("check_constraint_reflection",)
class AutogenCheckConstraintTest(AutogenFixtureTest, TestBase):
__backend__ = True
__requires__ = ("check_constraint_reflection",)
- def test_add_check_constraint(self):
+ def test_add_check_constraint(self, col_and_check):
m1 = MetaData()
m2 = MetaData()
m1 = MetaData()
m2 = MetaData()
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
)
diffs = self._fixture(m1, m2)
)
diffs = self._fixture(m1, m2)
eq_(diffs[0][0], "add_constraint")
eq_(diffs[0][1].name, "ck_t_x_positive")
eq_(diffs[0][0], "add_constraint")
eq_(diffs[0][1].name, "ck_t_x_positive")
- def test_can_be_disabled_via_exclusion(self):
+ def test_can_be_disabled_via_exclusion(self, col_and_check):
m1 = MetaData()
m2 = MetaData()
m1 = MetaData()
m2 = MetaData()
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
)
diffs = self._fixture(m1, m2, opts=_ck_plugin_disabled_opts)
)
diffs = self._fixture(m1, m2, opts=_ck_plugin_disabled_opts)
- def test_remove_check_constraint(self):
+ def test_remove_check_constraint(self, col_and_check):
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
eq_(diffs[0][0], "remove_constraint")
eq_(diffs[0][1].name, "ck_t_x_positive")
eq_(diffs[0][0], "remove_constraint")
eq_(diffs[0][1].name, "ck_t_x_positive")
- def test_same_name_different_expression_no_change(self):
+ def test_same_name_different_expression_no_change(self, col_and_check):
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
- Column("x", Integer),
- CheckConstraint("x > 5", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 5", name="ck_t_x_positive")
+ ),
)
diffs = self._fixture(m1, m2)
eq_(diffs, [])
)
diffs = self._fixture(m1, m2)
eq_(diffs, [])
- def test_compare_check_constraint_is_different(self, monkeypatch):
+ def test_compare_check_constraint_is_different(
+ self, col_and_check, monkeypatch
+ ):
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
- Column("x", Integer),
- CheckConstraint("x > 5", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 5", name="ck_t_x_positive")
+ ),
{"add_constraint", "remove_constraint"},
)
{"add_constraint", "remove_constraint"},
)
- def test_compare_check_constraint_is_skip(self, monkeypatch):
+ def test_compare_check_constraint_is_skip(
+ self, col_and_check, monkeypatch
+ ):
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
- Column("x", Integer),
- CheckConstraint("x > 5", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 5", name="ck_t_x_positive")
+ ),
- def test_no_change_check_constraint(self):
+ def test_no_change_check_constraint(self, col_and_check):
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
)
diffs = self._fixture(m1, m2)
eq_(diffs, [])
)
diffs = self._fixture(m1, m2)
eq_(diffs, [])
- def test_unnamed_check_constraint_in_metadata_ignored(self):
+ def test_unnamed_check_constraint_in_metadata_ignored(self, col_and_check):
m1 = MetaData()
m2 = MetaData()
m1 = MetaData()
m2 = MetaData()
- Column("x", Integer),
- CheckConstraint("x > 0"),
+ *col_and_check("x", Integer, CheckConstraint("x > 0")),
)
diffs = self._fixture(m1, m2)
)
diffs = self._fixture(m1, m2)
- def test_multiple_check_constraints(self):
+ def test_multiple_check_constraints(self, col_and_check):
m1 = MetaData()
m2 = MetaData()
m1 = MetaData()
m2 = MetaData()
"t",
m2,
Column("x", Integer),
"t",
m2,
Column("x", Integer),
CheckConstraint("x > 0", name="ck_x"),
CheckConstraint("x > 0", name="ck_x"),
- CheckConstraint("y > 0", name="ck_y"),
+ *col_and_check(
+ "y", Integer, CheckConstraint("y > 0", name="ck_y")
+ ),
)
diffs = self._fixture(m1, m2)
)
diffs = self._fixture(m1, m2)
eq_(diffs[0][0], "add_constraint")
eq_(diffs[0][1].name, "ck_y")
eq_(diffs[0][0], "add_constraint")
eq_(diffs[0][1].name, "ck_y")
- def test_remove_one_of_multiple(self):
+ def test_remove_one_of_multiple(self, col_and_check):
m1 = MetaData()
m2 = MetaData()
m1 = MetaData()
m2 = MetaData()
"t",
m1,
Column("x", Integer),
"t",
m1,
Column("x", Integer),
CheckConstraint("x > 0", name="ck_x"),
CheckConstraint("x > 0", name="ck_x"),
- CheckConstraint("y > 0", name="ck_y"),
+ *col_and_check(
+ "y", Integer, CheckConstraint("y > 0", name="ck_y")
+ ),
__backend__ = True
__requires__ = ("check_constraint_reflection",)
__backend__ = True
__requires__ = ("check_constraint_reflection",)
- def test_add_check_constraint_schema(self):
+ def test_add_check_constraint_schema(self, col_and_check):
m1 = MetaData()
m2 = MetaData()
m1 = MetaData()
m2 = MetaData()
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
schema=config.test_schema,
)
schema=config.test_schema,
)
eq_(diffs[0][1].name, "ck_t_x_positive")
eq_(diffs[0][1].table.schema, config.test_schema)
eq_(diffs[0][1].name, "ck_t_x_positive")
eq_(diffs[0][1].table.schema, config.test_schema)
- def test_remove_check_constraint_schema(self):
+ def test_remove_check_constraint_schema(self, col_and_check):
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
schema=config.test_schema,
)
schema=config.test_schema,
)
__backend__ = True
__requires__ = ("check_constraint_reflection",)
__backend__ = True
__requires__ = ("check_constraint_reflection",)
- def test_include_name_excludes_reflected_check_constraint(self):
+ def test_include_name_excludes_reflected_check_constraint(
+ self, col_and_check
+ ):
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
- def test_include_object_excludes_add(self):
+ def test_include_object_excludes_add(self, col_and_check):
m1 = MetaData()
m2 = MetaData()
m1 = MetaData()
m2 = MetaData()
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
)
def include_object(obj, name, type_, reflected, compare_to):
)
def include_object(obj, name, type_, reflected, compare_to):
- def test_include_object_excludes_remove(self):
+ def test_include_object_excludes_remove(self, col_and_check):
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
- def test_include_object_receives_correct_args_for_add(self):
+ def test_include_object_receives_correct_args_for_add(self, col_and_check):
m1 = MetaData()
m2 = MetaData()
m1 = MetaData()
m2 = MetaData()
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
eq_(calls[0][2], False)
eq_(calls[0][3], None)
eq_(calls[0][2], False)
eq_(calls[0][3], None)
- def test_include_object_receives_correct_args_for_remove(self):
+ def test_include_object_receives_correct_args_for_remove(
+ self, col_and_check
+ ):
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
eng.dialect.get_check_constraints = unimpl
eng.dialect.get_multi_check_constraints = unimpl
eng.dialect.get_check_constraints = unimpl
eng.dialect.get_multi_check_constraints = unimpl
- def test_no_reflection_graceful_skip_add(self):
+ def test_no_reflection_graceful_skip_add(self, col_and_check):
m1 = MetaData()
m2 = MetaData()
m1 = MetaData()
m2 = MetaData()
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
)
diffs = self._fixture(m1, m2)
)
diffs = self._fixture(m1, m2)
- def test_no_reflection_graceful_skip_remove(self):
+ def test_no_reflection_graceful_skip_remove(self, col_and_check):
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
m1 = MetaData()
m2 = MetaData()
Table(
"t",
m1,
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
__backend__ = True
__requires__ = ("check_constraint_reflection",)
__backend__ = True
__requires__ = ("check_constraint_reflection",)
- def test_add_named_via_convention(self):
+ def test_add_named_via_convention(self, col_and_check):
m1 = MetaData()
m2 = MetaData(
naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
m1 = MetaData()
m2 = MetaData(
naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
- Column("x", Integer),
- CheckConstraint("x > 0", name="x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="x_positive")
+ ),
)
diffs = self._fixture(m1, m2)
)
diffs = self._fixture(m1, m2)
eq_(diffs[0][0], "add_constraint")
eq_(diffs[0][1].name, "ck_t_x_positive")
eq_(diffs[0][0], "add_constraint")
eq_(diffs[0][1].name, "ck_t_x_positive")
- def test_remove_named_via_convention(self):
+ def test_remove_named_via_convention(self, col_and_check):
m1 = MetaData()
m2 = MetaData(
naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
m1 = MetaData()
m2 = MetaData(
naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
)
Table("t", m2, Column("x", Integer))
)
Table("t", m2, Column("x", Integer))
eq_(diffs[0][0], "remove_constraint")
eq_(diffs[0][1].name, "ck_t_x_positive")
eq_(diffs[0][0], "remove_constraint")
eq_(diffs[0][1].name, "ck_t_x_positive")
- def test_no_change_named_via_convention(self):
+ def test_no_change_named_via_convention(self, col_and_check):
m1 = MetaData()
m2 = MetaData(
naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
m1 = MetaData()
m2 = MetaData(
naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
- Column("x", Integer),
- CheckConstraint("x > 0", name="ck_t_x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="ck_t_x_positive")
+ ),
- Column("x", Integer),
- CheckConstraint("x > 0", name="x_positive"),
+ *col_and_check(
+ "x", Integer, CheckConstraint("x > 0", name="x_positive")
+ ),
)
diffs = self._fixture(m1, m2)
)
diffs = self._fixture(m1, m2)