ix = sa_schema.Index(
params["name"],
*[conn_table.c[cname] for cname in params["column_names"]],
- unique=params["unique"]
+ unique=params["unique"],
+ _table=conn_table
)
if "duplicates_constraint" in params:
ix.info["duplicates_constraint"] = params["duplicates_constraint"]
ops.AddConstraintOp.from_constraint(new.const)
)
- for added_name in sorted(set(metadata_names).difference(conn_names)):
- obj = metadata_names[added_name]
- obj_added(obj)
+ for removed_name in sorted(set(conn_names).difference(metadata_names)):
+ conn_obj = conn_names[removed_name]
+ if not conn_obj.is_index and conn_obj.sig in unnamed_metadata_uniques:
+ continue
+ elif removed_name in doubled_constraints:
+ if (
+ conn_obj.sig not in metadata_indexes_by_sig
+ and conn_obj.sig not in metadata_uniques_by_sig
+ ):
+ conn_uq, conn_idx = doubled_constraints[removed_name]
+ obj_removed(conn_uq)
+ obj_removed(conn_idx)
+ else:
+ obj_removed(conn_obj)
for existing_name in sorted(set(metadata_names).intersection(conn_names)):
metadata_obj = metadata_names[existing_name]
if msg:
obj_changed(conn_obj, metadata_obj, msg)
- for removed_name in sorted(set(conn_names).difference(metadata_names)):
- conn_obj = conn_names[removed_name]
- if not conn_obj.is_index and conn_obj.sig in unnamed_metadata_uniques:
- continue
- elif removed_name in doubled_constraints:
- if (
- conn_obj.sig not in metadata_indexes_by_sig
- and conn_obj.sig not in metadata_uniques_by_sig
- ):
- conn_uq, conn_idx = doubled_constraints[removed_name]
- obj_removed(conn_uq)
- obj_removed(conn_idx)
- else:
- obj_removed(conn_obj)
+ for added_name in sorted(set(metadata_names).difference(conn_names)):
+ obj = metadata_names[added_name]
+ obj_added(obj)
for uq_sig in unnamed_metadata_uniques:
if uq_sig not in conn_uniques_by_sig:
def doesnt_have_check_uq_constraints(config):
from sqlalchemy import inspect
- # temporary
- if config.db.name == "oracle":
- return True
-
insp = inspect(config.db)
try:
insp.get_unique_constraints("x")
--- /dev/null
+.. change::
+ :tags: bug, autogenerate
+ :tickets: 786
+
+ Changed the default ordering of "CREATE" and "DROP" statements indexes and
+ unique constraints within the autogenerate process, so that for example in
+ an upgrade() operation, a particular index or constraint that is to be
+ replaced such as for a casing convention change will not produce any naming
+ conflicts. For foreign key constraint objects, this is already how
+ constraints are ordered, and for table objects, users would normally want
+ to use :meth:`.Operations.rename_table` in any case.
def reflects_unique_constraints_unambiguously(self):
return exclusions.fails_on(["mysql", "mariadb", "oracle"])
+ @property
+ def reflects_indexes_w_sorting(self):
+ # TODO: figure out what's happening on the SQLAlchemy side
+ # when we reflect an index that has asc() / desc() on the column
+ return exclusions.fails_on(["oracle"])
+
+ @property
+ def long_names(self):
+ if sqla_compat.sqla_14:
+ return exclusions.skip_if("oracle<18")
+ else:
+ return exclusions.skip_if("oracle")
+
@property
def reflects_pk_names(self):
"""Target driver reflects the name of primary key constraints."""
eq_(diffs, [])
+ def test_casing_convention_changed_so_put_drops_first(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "some_table",
+ m1,
+ Column("test", String(10), primary_key=True),
+ mysql_engine="InnoDB",
+ )
+
+ Table(
+ "user",
+ m1,
+ Column("id", Integer, primary_key=True),
+ Column("name", String(50), nullable=False),
+ Column("a1", String(10), server_default="x"),
+ Column("test2", String(10)),
+ ForeignKeyConstraint(["test2"], ["some_table.test"], name="MyFK"),
+ mysql_engine="InnoDB",
+ )
+
+ Table(
+ "some_table",
+ m2,
+ Column("test", String(10), primary_key=True),
+ mysql_engine="InnoDB",
+ )
+
+ # foreign key autogen currently does not take "name" into account,
+ # so change the def just for the purposes of testing the
+ # add/drop order for now.
+ Table(
+ "user",
+ m2,
+ Column("id", Integer, primary_key=True),
+ Column("name", String(50), nullable=False),
+ Column("a1", String(10), server_default="x"),
+ Column("test2", String(10)),
+ ForeignKeyConstraint(["a1"], ["some_table.test"], name="myfk"),
+ mysql_engine="InnoDB",
+ )
+
+ diffs = self._fixture(m1, m2)
+
+ self._assert_fk_diff(
+ diffs[0],
+ "remove_fk",
+ "user",
+ ["test2"],
+ "some_table",
+ ["test"],
+ name="MyFK" if config.requirements.fk_names.enabled else None,
+ )
+
+ self._assert_fk_diff(
+ diffs[1],
+ "add_fk",
+ "user",
+ ["a1"],
+ "some_table",
+ ["test"],
+ name="myfk",
+ )
+
def test_add_composite_fk_with_name(self):
m1 = MetaData()
m2 = MetaData()
diffs = self._fixture(m1, m2)
if self.reports_unique_constraints:
- eq_(diffs[0][0], "add_constraint")
- eq_(diffs[0][1].name, "uq_user_name")
+ eq_(diffs[0][0], "remove_index")
+ eq_(diffs[0][1].name, "ix_user_name")
+
+ eq_(diffs[1][0], "add_constraint")
+ eq_(diffs[1][1].name, "uq_user_name")
- eq_(diffs[1][0], "remove_index")
- eq_(diffs[1][1].name, "ix_user_name")
else:
eq_(diffs[0][0], "remove_index")
eq_(diffs[0][1].name, "ix_user_name")
diffs = self._fixture(m1, m2, max_identifier_length=30)
eq_(diffs, [])
+ @config.requirements.long_names
def test_nothing_ix_changed_labels_were_truncated(self):
m1 = MetaData(
naming_convention={
diffs = self._fixture(m1, m2, max_identifier_length=30)
eq_(diffs, [])
+ @config.requirements.long_names
def test_nothing_changed_uq_w_mixed_case_nconv_name(self):
m1 = MetaData(
naming_convention={
diffs = self._fixture(m1, m2)
eq_(diffs, [])
+ @config.requirements.long_names
def test_nothing_changed_ix_w_mixed_case_nconv_name(self):
m1 = MetaData(
naming_convention={
diffs = self._fixture(m1, m2)
eq_(diffs, [])
+ def test_ix_casing_convention_changed_so_put_drops_first(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ ix1 = Index("SomeCasingConvention", "x")
+ Table(
+ "new_idx",
+ m1,
+ Column("id1", Integer, primary_key=True),
+ Column("x", String(20)),
+ ix1,
+ )
+
+ ix2 = Index("somecasingconvention", "x")
+ Table(
+ "new_idx",
+ m2,
+ Column("id1", Integer, primary_key=True),
+ Column("x", String(20)),
+ ix2,
+ )
+
+ diffs = self._fixture(m1, m2)
+
+ eq_(
+ [(d[0], d[1].name) for d in diffs],
+ [
+ ("remove_index", "SomeCasingConvention"),
+ ("add_index", "somecasingconvention"),
+ ],
+ )
+
+ def test_uq_casing_convention_changed_so_put_drops_first(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ uq1 = UniqueConstraint("x", name="SomeCasingConvention")
+ Table(
+ "new_idx",
+ m1,
+ Column("id1", Integer, primary_key=True),
+ Column("x", String(20)),
+ uq1,
+ )
+
+ uq2 = UniqueConstraint("x", name="somecasingconvention")
+ Table(
+ "new_idx",
+ m2,
+ Column("id1", Integer, primary_key=True),
+ Column("x", String(20)),
+ uq2,
+ )
+
+ diffs = self._fixture(m1, m2)
+
+ if self.reports_unique_constraints_as_indexes:
+ eq_(
+ [(d[0], d[1].name) for d in diffs],
+ [
+ ("remove_index", "SomeCasingConvention"),
+ ("add_constraint", "somecasingconvention"),
+ ],
+ )
+ else:
+ eq_(
+ [(d[0], d[1].name) for d in diffs],
+ [
+ ("remove_constraint", "SomeCasingConvention"),
+ ("add_constraint", "somecasingconvention"),
+ ],
+ )
+
def test_new_idx_index_named_as_column(self):
m1 = MetaData()
m2 = MetaData()
diffs = self._fixture(m1, m2)
diffs = set(
- (cmd, ("x" in obj.name) if obj.name is not None else False)
+ (
+ cmd,
+ isinstance(obj, (UniqueConstraint, Index))
+ if obj.name is not None
+ else False,
+ )
for cmd, obj in diffs
)
if self.reports_unnamed_constraints:
eq_(diffs[0][0], "add_index")
+ @config.requirements.reflects_indexes_w_sorting
def test_unchanged_idx_non_col(self):
m1 = MetaData()
m2 = MetaData()
reports_unique_constraints = False
__only_on__ = "sqlite"
+ def test_uq_casing_convention_changed_so_put_drops_first(self):
+ config.skip_test(
+ "unique constraint reflection disabled for this suite"
+ )
+
def test_unique_not_reported(self):
m1 = MetaData()
Table(
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
