self.is_unique = bool(const.unique)
def md_name_to_sql_name(self, context):
- return sqla_compat._get_index_final_name(context.dialect, self.const)
+ return sqla_compat._get_constraint_final_name(
+ self.const, context.dialect
+ )
@property
def column_names(self):
conn_indexes,
metadata_unique_constraints,
metadata_indexes,
+ autogen_context.dialect,
)
# 3. give the dialect a chance to omit indexes and constraints that
conn_uniques_by_name = dict((c.name, c) for c in conn_unique_constraints)
conn_indexes_by_name = dict((c.name, c) for c in conn_indexes)
-
conn_names = dict(
(c.name, c)
for c in conn_unique_constraints.union(conn_indexes)
conn_indexes,
metadata_unique_constraints,
metadata_indexes,
+ dialect,
):
# dedupe unique indexes vs. constraints, since MySQL / Oracle
# doesn't really have unique constraints as a separate construct.
# that already seem to be defined one way or the other
# on that side. This logic was formerly local to MySQL dialect,
# generalized to Oracle and others. See #276
+
+ # resolve final rendered name for unique constraints defined in the
+ # metadata. this includes truncation of long names. naming convention
+ # names currently should already be set as cons.name, however leave this
+ # to the sqla_compat to decide.
+ metadata_cons_names = [
+ (sqla_compat._get_constraint_final_name(cons, dialect), cons)
+ for cons in metadata_unique_constraints
+ ]
+
metadata_uq_names = set(
- [
- cons.name
- for cons in metadata_unique_constraints
- if cons.name is not None
- ]
+ name for name, cons in metadata_cons_names if name is not None
)
unnamed_metadata_uqs = set(
[
_uq_constraint_sig(cons).sig
- for cons in metadata_unique_constraints
- if cons.name is None
+ for name, cons in metadata_cons_names
+ if name is None
]
)
metadata_ix_names = set(
- [cons.name for cons in metadata_indexes if cons.unique]
+ [
+ sqla_compat._get_constraint_final_name(cons, dialect)
+ for cons in metadata_indexes
+ if cons.unique
+ ]
)
+
+ # for reflection side, names are in their final database form
+ # already since they're from the database
conn_ix_names = dict(
(cons.name, cons) for cons in conn_indexes if cons.unique
)
if idx.name in removed:
metadata_indexes.remove(idx)
- def _legacy_correct_for_dupe_uq_uix(
- self,
- conn_unique_constraints,
- conn_indexes,
- metadata_unique_constraints,
- metadata_indexes,
- ):
-
- # then dedupe unique indexes vs. constraints, since MySQL
- # doesn't really have unique constraints as a separate construct.
- # but look in the metadata and try to maintain constructs
- # that already seem to be defined one way or the other
- # on that side. See #276
- metadata_uq_names = set(
- [
- cons.name
- for cons in metadata_unique_constraints
- if cons.name is not None
- ]
- )
-
- unnamed_metadata_uqs = set(
- [
- compare._uq_constraint_sig(cons).sig
- for cons in metadata_unique_constraints
- if cons.name is None
- ]
- )
-
- metadata_ix_names = set(
- [cons.name for cons in metadata_indexes if cons.unique]
- )
- conn_uq_names = dict(
- (cons.name, cons) for cons in conn_unique_constraints
- )
- conn_ix_names = dict(
- (cons.name, cons) for cons in conn_indexes if cons.unique
- )
-
- for overlap in set(conn_uq_names).intersection(conn_ix_names):
- if overlap not in metadata_uq_names:
- if (
- compare._uq_constraint_sig(conn_uq_names[overlap]).sig
- not in unnamed_metadata_uqs
- ):
-
- conn_unique_constraints.discard(conn_uq_names[overlap])
- elif overlap not in metadata_ix_names:
- conn_indexes.discard(conn_ix_names[overlap])
-
def correct_for_autogen_foreignkeys(self, conn_fks, metadata_fks):
conn_fk_by_sig = dict(
(compare._fk_constraint_sig(fk).sig, fk) for fk in conn_fks
return {}
-def _get_index_final_name(dialect, idx):
- if sqla_14:
- return _get_constraint_final_name(idx, dialect)
-
- # prior to SQLAlchemy 1.4, work around quoting logic to get at the
- # final compiled name without quotes.
- if hasattr(idx.name, "quote"):
- # might be quoted_name, might be truncated_name, keep it the
- # same
- quoted_name_cls = type(idx.name)
- new_name = quoted_name_cls(str(idx.name), quote=False)
- idx = schema.Index(name=new_name)
- return dialect.ddl_compiler(dialect, None)._prepared_index_name(idx)
-
-
def _get_constraint_final_name(constraint, dialect):
- if sqla_14:
- if constraint.name is None:
- return None
-
+ if constraint.name is None:
+ return None
+ elif sqla_14:
# for SQLAlchemy 1.4 we would like to have the option to expand
# the use of "deferred" names for constraints as well as to have
# some flexibility with "None" name and similar; make use of new
constraint, _alembic_quote=False
)
else:
- return constraint.name
+
+ # prior to SQLAlchemy 1.4, work around quoting logic to get at the
+ # final compiled name without quotes.
+ if hasattr(constraint.name, "quote"):
+ # might be quoted_name, might be truncated_name, keep it the
+ # same
+ quoted_name_cls = type(constraint.name)
+ new_name = quoted_name_cls(str(constraint.name), quote=False)
+ constraint = constraint.__class__(name=new_name)
+
+ if isinstance(constraint, schema.Index):
+ return dialect.ddl_compiler(dialect, None)._prepared_index_name(
+ constraint
+ )
+ else:
+ return dialect.identifier_preparer.format_constraint(constraint)
def _constraint_is_named(constraint, dialect):
--- /dev/null
+.. change::
+ :tags: bug, autogenerate
+ :tickets: 647
+
+ Adjusted the unique constraint comparison logic in a similar manner as that
+ of :ticket:`421` did for indexes in order to take into account SQLAlchemy's
+ own truncation of long constraint names when a naming convention is in use.
+ Without this step, a name that is truncated by SQLAlchemy based on a unique
+ constraint naming convention or hardcoded name will not compare properly.
+
opts=None,
object_filters=_default_object_filters,
return_ops=False,
+ max_identifier_length=None,
):
- self.metadata, model_metadata = m1, m2
- for m in util.to_list(self.metadata):
- m.create_all(self.bind)
-
- with self.bind.connect() as conn:
- ctx_opts = {
- "compare_type": True,
- "compare_server_default": True,
- "target_metadata": model_metadata,
- "upgrade_token": "upgrades",
- "downgrade_token": "downgrades",
- "alembic_module_prefix": "op.",
- "sqlalchemy_module_prefix": "sa.",
- "include_object": object_filters,
- "include_schemas": include_schemas,
- }
- if opts:
- ctx_opts.update(opts)
- self.context = context = MigrationContext.configure(
- connection=conn, opts=ctx_opts
- )
-
- autogen_context = api.AutogenContext(context, model_metadata)
- uo = ops.UpgradeOps(ops=[])
- autogenerate._produce_net_changes(autogen_context, uo)
-
- if return_ops:
- return uo
- else:
- return uo.as_diffs()
+
+ if max_identifier_length:
+ dialect = self.bind.dialect
+ existing_length = dialect.max_identifier_length
+ dialect.max_identifier_length = (
+ dialect._user_defined_max_identifier_length
+ ) = max_identifier_length
+ try:
+ self.metadata, model_metadata = m1, m2
+ for m in util.to_list(self.metadata):
+ m.create_all(self.bind)
+
+ with self.bind.connect() as conn:
+ ctx_opts = {
+ "compare_type": True,
+ "compare_server_default": True,
+ "target_metadata": model_metadata,
+ "upgrade_token": "upgrades",
+ "downgrade_token": "downgrades",
+ "alembic_module_prefix": "op.",
+ "sqlalchemy_module_prefix": "sa.",
+ "include_object": object_filters,
+ "include_schemas": include_schemas,
+ }
+ if opts:
+ ctx_opts.update(opts)
+ self.context = context = MigrationContext.configure(
+ connection=conn, opts=ctx_opts
+ )
+
+ autogen_context = api.AutogenContext(context, model_metadata)
+ uo = ops.UpgradeOps(ops=[])
+ autogenerate._produce_net_changes(autogen_context, uo)
+
+ if return_ops:
+ return uo
+ else:
+ return uo.as_diffs()
+ finally:
+ if max_identifier_length:
+ dialect = self.bind.dialect
+ dialect.max_identifier_length = (
+ dialect._user_defined_max_identifier_length
+ ) = existing_length
reports_unnamed_constraints = False
diffs = self._fixture(m1, m2)
eq_(diffs, [])
+ def test_nothing_uq_changed_labels_were_truncated(self):
+ m1 = MetaData(
+ naming_convention={
+ "ix": "index_%(table_name)s_%(column_0_label)s",
+ "uq": "unique_%(table_name)s_%(column_0_label)s",
+ }
+ )
+ m2 = MetaData(
+ naming_convention={
+ "ix": "index_%(table_name)s_%(column_0_label)s",
+ "uq": "unique_%(table_name)s_%(column_0_label)s",
+ }
+ )
+
+ Table(
+ "nothing_changed",
+ m1,
+ Column("id1", Integer, primary_key=True),
+ Column("id2", Integer, primary_key=True),
+ Column("a_long_name", String(20), unique=True),
+ mysql_engine="InnoDB",
+ )
+
+ Table(
+ "nothing_changed",
+ m2,
+ Column("id1", Integer, primary_key=True),
+ Column("id2", Integer, primary_key=True),
+ Column("a_long_name", String(20), unique=True),
+ mysql_engine="InnoDB",
+ )
+ diffs = self._fixture(m1, m2, max_identifier_length=30)
+ eq_(diffs, [])
+
+ def test_nothing_ix_changed_labels_were_truncated(self):
+ m1 = MetaData(
+ naming_convention={
+ "ix": "index_%(table_name)s_%(column_0_label)s",
+ "uq": "unique_%(table_name)s_%(column_0_label)s",
+ }
+ )
+ m2 = MetaData(
+ naming_convention={
+ "ix": "index_%(table_name)s_%(column_0_label)s",
+ "uq": "unique_%(table_name)s_%(column_0_label)s",
+ }
+ )
+
+ Table(
+ "nothing_changed",
+ m1,
+ Column("id1", Integer, primary_key=True),
+ Column("id2", Integer, primary_key=True),
+ Column("a_particularly_long_column_name", String(20), index=True),
+ mysql_engine="InnoDB",
+ )
+
+ Table(
+ "nothing_changed",
+ m2,
+ Column("id1", Integer, primary_key=True),
+ Column("id2", Integer, primary_key=True),
+ Column("a_particularly_long_column_name", String(20), index=True),
+ mysql_engine="InnoDB",
+ )
+ diffs = self._fixture(m1, m2, max_identifier_length=30)
+ eq_(diffs, [])
+
def test_nothing_changed_two(self):
m1 = MetaData()
m2 = MetaData()
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
