def _compare_indexes_and_uniques(schema, tname, object_filters, conn_table,
metadata_table, diffs, autogen_context, inspector):
+ is_create_table = conn_table is None
+
# 1a. get raw indexes and unique constraints from metadata ...
metadata_unique_constraints = set(uq for uq in metadata_table.constraints
if isinstance(uq, sa_schema.UniqueConstraint)
])
)
else:
+ if is_create_table:
+ # unique constraints are created inline with table defs
+ return
diffs.append(("add_constraint", obj.const))
log.info("Detected added unique constraint '%s' on %s",
obj.name, ', '.join([
"""
from .compare import _get_index_column_names
- text = "op.create_index('%(name)s', '%(table)s', %(columns)s, unique=%(unique)r%(schema)s%(kwargs)s)" % {
+ text = "%(prefix)screate_index('%(name)s', '%(table)s', %(columns)s, "\
+ "unique=%(unique)r%(schema)s%(kwargs)s)" % {
+ 'prefix': _alembic_autogenerate_prefix(autogen_context),
'name': index.name,
'table': index.table,
'columns': _get_index_column_names(index),
Generate Alembic operations for the DROP INDEX of an
:class:`~sqlalchemy.schema.Index` instance.
"""
- text = "op.drop_index('%s', '%s')" % (index.name, index.table)
+ text = "%sdrop_index('%s', '%s')" % (
+ _alembic_autogenerate_prefix(autogen_context),
+ index.name,
+ index.table)
return text
+def _render_unique_constraint(constraint, autogen_context):
+ rendered = _user_defined_render("unique", constraint, autogen_context)
+ if rendered is not False:
+ return rendered
+
+ return _uq_constraint(constraint, autogen_context, False)
+
+
def _add_unique_constraint(constraint, autogen_context):
"""
Generate Alembic operations for the ALTER TABLE .. ADD CONSTRAINT ...
UNIQUE of a :class:`~sqlalchemy.schema.UniqueConstraint` instance.
"""
- text = "%(prefix)screate_unique_constraint(%(name)r, '%(table)s', %(columns)s"\
- "%(deferrable)s%(initially)s%(schema)s)" % {
- 'prefix': _alembic_autogenerate_prefix(autogen_context),
- 'name': constraint.name,
- 'table': constraint.table,
- 'columns': [col.name for col in constraint.columns],
- 'deferrable': (", deferrable='%s'" % constraint.deferrable) if constraint.deferrable else '',
- 'initially': (", initially='%s'" % constraint.initially) if constraint.initially else '',
- 'schema': (", schema='%s'" % constraint.table.schema) if constraint.table.schema else ''
+ return _uq_constraint(constraint, autogen_context, True)
+
+def _uq_constraint(constraint, autogen_context, alter):
+ opts = []
+ if constraint.deferrable:
+ opts.append(("deferrable", str(constraint.deferrable)))
+ if constraint.initially:
+ opts.append(("initially", str(constraint.initially)))
+ if alter and constraint.table.schema:
+ opts.append(("schema", str(constraint.table.schema)))
+ if not alter and constraint.name:
+ opts.append(("name", constraint.name))
+
+ if alter:
+ args = [repr(constraint.name), repr(constraint.table.name)]
+ args.append(repr([col.name for col in constraint.columns]))
+ args.extend(["%s=%r" % (k, v) for k, v in opts])
+ return "%(prefix)screate_unique_constraint(%(args)s)" % {
+ 'prefix': _alembic_autogenerate_prefix(autogen_context),
+ 'args': ", ".join(args)
+ }
+ else:
+ args = [repr(col.name) for col in constraint.columns]
+ args.extend(["%s=%r" % (k, v) for k, v in opts])
+ return "%(prefix)sUniqueConstraint(%(args)s)" % {
+ "prefix": _sqlalchemy_autogenerate_prefix(autogen_context),
+ "args": ", ".join(args)
}
- return text
+
def _add_fk_constraint(constraint, autogen_context):
raise NotImplementedError()
)
}
-def _render_unique_constraint(constraint, autogen_context):
- rendered = _user_defined_render("unique", constraint, autogen_context)
- if rendered is not False:
- return rendered
-
- opts = []
- if constraint.name:
- opts.append(("name", "'%s'" % constraint.name))
- return "%(prefix)sUniqueConstraint(%(cols)s%(opts)s)" % {
- 'opts': ", " + (", ".join("%s=%s" % (k, v)
- for k, v in opts)) if opts else "",
- 'cols': ",".join(["'%s'" % c.name for c in constraint.columns]),
- "prefix": _sqlalchemy_autogenerate_prefix(autogen_context)
- }
_constraint_renderers = {
sa_schema.PrimaryKeyConstraint: _render_primary_key,
sa_schema.ForeignKeyConstraint: _render_foreign_key,
currently report on constraints that were made with this technique,
hence they'd come out as "added" on every run.
+ * the ``op.create_table()`` directive will be auto-generated with
+ the ``UniqueConstraint`` objects inline, but will not double them
+ up with a separate ``create_unique_constraint()`` call, which may
+ have been occurring. Indexes still get rendered as distinct
+ ``op.create_index()`` calls even when the corresponding table was
+ created in the same script.
+
+ * the inline ``UniqueConstraint`` within ``op.create_table()`` includes
+ all the options like ``deferrable``, ``initially``, etc. Previously
+ these weren't rendering.
+
.. change::
:tags: feature, mssql
diffs = ((cmd, obj.name) for cmd, obj in diffs)
assert ("remove_constraint", "xidx") in diffs
+ def test_dont_add_uq_on_table_create(self):
+ m1 = MetaData()
+ m2 = MetaData()
+ Table('no_uq', m2, Column('x', String(50), unique=True))
+ diffs = self._fixture(m1, m2)
+
+ eq_(diffs[0][0], "add_table")
+ eq_(len(diffs), 1)
+ assert UniqueConstraint in set(type(c) for c in diffs[0][1].constraints)
+
+ def test_add_uq_ix_on_table_create(self):
+ m1 = MetaData()
+ m2 = MetaData()
+ Table('add_ix', m2, Column('x', String(50), unique=True, index=True))
+ diffs = self._fixture(m1, m2)
+
+ eq_(diffs[0][0], "add_table")
+ eq_(len(diffs), 2)
+ assert UniqueConstraint not in set(type(c) for c in diffs[0][1].constraints)
+ eq_(diffs[1][0], "add_index")
+ eq_(diffs[1][1].unique, True)
+
+ def test_add_ix_on_table_create(self):
+ m1 = MetaData()
+ m2 = MetaData()
+ Table('add_ix', m2, Column('x', String(50), index=True))
+ diffs = self._fixture(m1, m2)
+
+ eq_(diffs[0][0], "add_table")
+ eq_(len(diffs), 2)
+ assert UniqueConstraint not in set(type(c) for c in diffs[0][1].constraints)
+ eq_(diffs[1][0], "add_index")
+ eq_(diffs[1][1].unique, False)
+
def _fixture(self, m1, m2):
self.metadata, model_metadata = m1, m2
self.metadata.create_all(self.bind)
- conn = self.bind.connect()
- self.context = context = MigrationContext.configure(
- connection=conn,
- 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.',
- }
- )
+ with self.bind.connect() as conn:
+ self.context = context = MigrationContext.configure(
+ connection=conn,
+ 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.',
+ }
+ )
- connection = context.bind
- autogen_context = {
- 'imports': set(),
- 'connection': connection,
- 'dialect': connection.dialect,
- 'context': context
- }
- diffs = []
- autogenerate._produce_net_changes(connection, model_metadata, diffs,
- autogen_context,
- object_filters=_default_object_filters,
- )
- return diffs
+ connection = context.bind
+ autogen_context = {
+ 'imports': set(),
+ 'connection': connection,
+ 'dialect': connection.dialect,
+ 'context': context
+ }
+ diffs = []
+ autogenerate._produce_net_changes(connection, model_metadata, diffs,
+ autogen_context,
+ object_filters=_default_object_filters,
+ )
+ return diffs
reports_unnamed_constraints = False
"sa.CheckConstraint('c > 5 AND c < 10')"
)
+ def test_render_unique_constraint_opts(self):
+ m = MetaData()
+ t = Table('t', m, Column('c', Integer))
+ eq_ignore_whitespace(
+ autogenerate.render._render_unique_constraint(
+ UniqueConstraint(t.c.c, name='uq_1', deferrable='XYZ'),
+ self.autogen_context
+ ),
+ "sa.UniqueConstraint('c', deferrable='XYZ', name='uq_1')"
+ )
+
def test_render_modify_nullable_w_default(self):
eq_ignore_whitespace(
autogenerate.render._modify_col(
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
