ensure conv is applied to all reflected index and constraint names

The autogenerate process will now apply the :meth:`.Operations.f` modifier
to the names of all constraints and indexes that are reflected from the
target database when generating migrations, which has the effect that these
names will not have any subsequent naming conventions applied to them when
the migration operations proceed.  As reflected objects already include the
exact name that's present in the database, these names should not be
modified.   The fix repairs the issue when using custom naming conventions
which feature the ``%(constraint_name)s`` token would cause names to be
double-processed, leading to errors in migration runs.

Fixes: #264
Change-Id: Ie48f65c41e66f3698cafe558cafa189f524c8cdf

diff --git a/alembic/autogenerate/compare.py b/alembic/autogenerate/compare.py
index 8d6d8f1b35f7ce4ad85b7d6b206d644f5b49a850..66e31ca46cad6a2b664f580fe7513ecb273e3b94 100644 (file)
--- a/alembic/autogenerate/compare.py
+++ b/alembic/autogenerate/compare.py
@@ -24,6 +24,7 @@ from sqlalchemy import schema as sa_schema
 from sqlalchemy import text
 from sqlalchemy import types as sqltypes
 from sqlalchemy.sql import expression
+from sqlalchemy.sql.elements import conv
 from sqlalchemy.sql.schema import ForeignKeyConstraint
 from sqlalchemy.sql.schema import Index
 from sqlalchemy.sql.schema import UniqueConstraint
@@ -216,7 +217,7 @@ def _compare_tables(
                 (inspector),
                 # fmt: on
             )
-            inspector.reflect_table(t, include_columns=None)
+            _InspectorConv(inspector).reflect_table(t, include_columns=None)
         if autogen_context.run_object_filters(t, tname, "table", True, None):
             modify_table_ops = ops.ModifyTableOps(tname, [], schema=s)
 
@@ -246,7 +247,8 @@ def _compare_tables(
                 _compat_autogen_column_reflect(inspector),
                 # fmt: on
             )
-            inspector.reflect_table(t, include_columns=None)
+            _InspectorConv(inspector).reflect_table(t, include_columns=None)
+
         conn_column_info[(s, tname)] = t
 
     for s, tname in sorted(existing_tables, key=lambda x: (x[0] or "", x[1])):
@@ -438,6 +440,55 @@ def _compare_columns(
 _C = TypeVar("_C", bound=Union[UniqueConstraint, ForeignKeyConstraint, Index])
 
 
+class _InspectorConv:
+    __slots__ = ("inspector",)
+
+    def __init__(self, inspector):
+        self.inspector = inspector
+
+    def _apply_reflectinfo_conv(self, consts):
+        if not consts:
+            return consts
+        for const in consts:
+            if const["name"] is not None and not isinstance(
+                const["name"], conv
+            ):
+                const["name"] = conv(const["name"])
+        return consts
+
+    def _apply_constraint_conv(self, consts):
+        if not consts:
+            return consts
+        for const in consts:
+            if const.name is not None and not isinstance(const.name, conv):
+                const.name = conv(const.name)
+        return consts
+
+    def get_indexes(self, *args, **kw):
+        return self._apply_reflectinfo_conv(
+            self.inspector.get_indexes(*args, **kw)
+        )
+
+    def get_unique_constraints(self, *args, **kw):
+        return self._apply_reflectinfo_conv(
+            self.inspector.get_unique_constraints(*args, **kw)
+        )
+
+    def get_foreign_keys(self, *args, **kw):
+        return self._apply_reflectinfo_conv(
+            self.inspector.get_foreign_keys(*args, **kw)
+        )
+
+    def reflect_table(self, table, *, include_columns):
+        self.inspector.reflect_table(table, include_columns=include_columns)
+
+        # I had a cool version of this using _ReflectInfo, however that doesn't
+        # work in 1.4 and it's not public API in 2.x.  Then this is just a two
+        # liner.  So there's no competition...
+        self._apply_constraint_conv(table.constraints)
+        self._apply_constraint_conv(table.indexes)
+
+
 @comparators.dispatch_for("table")
 def _compare_indexes_and_uniques(
     autogen_context: AutogenContext,
@@ -473,9 +524,10 @@ def _compare_indexes_and_uniques(
     if conn_table is not None:
         # 1b. ... and from connection, if the table exists
         try:
-            conn_uniques = inspector.get_unique_constraints(  # type:ignore[assignment] # noqa
+            conn_uniques = _InspectorConv(inspector).get_unique_constraints(
                 tname, schema=schema
             )
+
             supports_unique_constraints = True
         except NotImplementedError:
             pass
@@ -498,7 +550,7 @@ def _compare_indexes_and_uniques(
                 if uq.get("duplicates_index"):
                     unique_constraints_duplicate_unique_indexes = True
         try:
-            conn_indexes = inspector.get_indexes(  # type:ignore[assignment]
+            conn_indexes = _InspectorConv(inspector).get_indexes(
                 tname, schema=schema
             )
         except NotImplementedError:
@@ -1178,7 +1230,9 @@ def _compare_foreign_keys(
 
     conn_fks_list = [
         fk
-        for fk in inspector.get_foreign_keys(tname, schema=schema)
+        for fk in _InspectorConv(inspector).get_foreign_keys(
+            tname, schema=schema
+        )
         if autogen_context.run_name_filters(
             fk["name"],
             "foreign_key_constraint",
@@ -1187,8 +1241,7 @@ def _compare_foreign_keys(
     ]
 
     conn_fks = {
-        _make_foreign_key(const, conn_table)  # type: ignore[arg-type]
-        for const in conn_fks_list
+        _make_foreign_key(const, conn_table) for const in conn_fks_list
     }
 
     impl = autogen_context.migration_context.impl

diff --git a/alembic/testing/suite/_autogen_fixtures.py b/alembic/testing/suite/_autogen_fixtures.py
index d838ebef1068b5cc38a4a18f1b16b6cd00876581..ed4acb26aec0344cff23d7df07c0a9294e6bc49c 100644 (file)
--- a/alembic/testing/suite/_autogen_fixtures.py
+++ b/alembic/testing/suite/_autogen_fixtures.py
@@ -14,6 +14,7 @@ from sqlalchemy import inspect
 from sqlalchemy import Integer
 from sqlalchemy import MetaData
 from sqlalchemy import Numeric
+from sqlalchemy import PrimaryKeyConstraint
 from sqlalchemy import String
 from sqlalchemy import Table
 from sqlalchemy import Text
@@ -149,6 +150,118 @@ class ModelOne:
         return m
 
 
+class NamingConvModel:
+    __requires__ = ("unique_constraint_reflection",)
+    configure_opts = {"conv_all_constraint_names": True}
+    naming_convention = {
+        "ix": "ix_%(column_0_label)s",
+        "uq": "uq_%(table_name)s_%(constraint_name)s",
+        "ck": "ck_%(table_name)s_%(constraint_name)s",
+        "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
+        "pk": "pk_%(table_name)s",
+    }
+
+    @classmethod
+    def _get_db_schema(cls):
+        # database side - assume all constraints have a name that
+        # we would assume here is a "db generated" name.  need to make
+        # sure these all render with op.f().
+        m = MetaData()
+        Table(
+            "x1",
+            m,
+            Column("q", Integer),
+            Index("db_x1_index_q", "q"),
+            PrimaryKeyConstraint("q", name="db_x1_primary_q"),
+        )
+        Table(
+            "x2",
+            m,
+            Column("q", Integer),
+            Column("p", ForeignKey("x1.q", name="db_x2_foreign_q")),
+            CheckConstraint("q > 5", name="db_x2_check_q"),
+        )
+        Table(
+            "x3",
+            m,
+            Column("q", Integer),
+            Column("r", Integer),
+            Column("s", Integer),
+            UniqueConstraint("q", name="db_x3_unique_q"),
+        )
+        Table(
+            "x4",
+            m,
+            Column("q", Integer),
+            PrimaryKeyConstraint("q", name="db_x4_primary_q"),
+        )
+        Table(
+            "x5",
+            m,
+            Column("q", Integer),
+            Column("p", ForeignKey("x4.q", name="db_x5_foreign_q")),
+            Column("r", Integer),
+            Column("s", Integer),
+            PrimaryKeyConstraint("q", name="db_x5_primary_q"),
+            UniqueConstraint("r", name="db_x5_unique_r"),
+            CheckConstraint("s > 5", name="db_x5_check_s"),
+        )
+        # SQLite and it's "no names needed" thing.  bleh.
+        # we can't have a name for these so you'll see "None" for the name.
+        Table(
+            "unnamed_sqlite",
+            m,
+            Column("q", Integer),
+            Column("r", Integer),
+            PrimaryKeyConstraint("q"),
+            UniqueConstraint("r"),
+        )
+        return m
+
+    @classmethod
+    def _get_model_schema(cls):
+        from sqlalchemy.sql.naming import conv
+
+        m = MetaData(naming_convention=cls.naming_convention)
+        Table(
+            "x1", m, Column("q", Integer, primary_key=True), Index(None, "q")
+        )
+        Table(
+            "x2",
+            m,
+            Column("q", Integer),
+            Column("p", ForeignKey("x1.q")),
+            CheckConstraint("q > 5", name="token_x2check1"),
+        )
+        Table(
+            "x3",
+            m,
+            Column("q", Integer),
+            Column("r", Integer),
+            Column("s", Integer),
+            UniqueConstraint("r", name="token_x3r"),
+            UniqueConstraint("s", name=conv("userdef_x3_unique_s")),
+        )
+        Table(
+            "x4",
+            m,
+            Column("q", Integer, primary_key=True),
+            Index("userdef_x4_idx_q", "q"),
+        )
+        Table(
+            "x6",
+            m,
+            Column("q", Integer, primary_key=True),
+            Column("p", ForeignKey("x4.q")),
+            Column("r", Integer),
+            Column("s", Integer),
+            UniqueConstraint("r", name="token_x6r"),
+            CheckConstraint("s > 5", "token_x6check1"),
+            CheckConstraint("s < 20", conv("userdef_x6_check_s")),
+        )
+        return m
+
+
 class _ComparesFKs:
     def _assert_fk_diff(
         self,

diff --git a/docs/build/unreleased/264.rst b/docs/build/unreleased/264.rst
new file mode 100644 (file)
index 0000000..bc1675b
--- /dev/null
+++ b/docs/build/unreleased/264.rst
@@ -0,0 +1,15 @@
+.. change::
+    :tags: bug, autogenerate
+    :tickets: 264
+
+    The autogenerate process will now apply the :meth:`.Operations.f` modifier
+    to the names of all constraints and indexes that are reflected from the
+    target database when generating migrations, which has the effect that these
+    names will not have any subsequent naming conventions applied to them when
+    the migration operations proceed.  As reflected objects already include the
+    exact name that's present in the database, these names should not be
+    modified.   The fix repairs the issue when using custom naming conventions
+    which feature the ``%(constraint_name)s`` token would cause names to be
+    double-processed, leading to errors in migration runs.
+
+

diff --git a/tests/test_autogen_composition.py b/tests/test_autogen_composition.py
index b4b5bf33353bd848179436b867593e7a39eb494c..e8688b84f7f7e416fda130d0377154980f40e5ab 100644 (file)
--- a/tests/test_autogen_composition.py
+++ b/tests/test_autogen_composition.py
@@ -12,6 +12,7 @@ from alembic.testing import TestBase
 from alembic.testing.suite._autogen_fixtures import _default_include_object
 from alembic.testing.suite._autogen_fixtures import AutogenTest
 from alembic.testing.suite._autogen_fixtures import ModelOne
+from alembic.testing.suite._autogen_fixtures import NamingConvModel
 
 
 class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
@@ -92,8 +93,7 @@ class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
     sa.PrimaryKeyConstraint('id')
     )
     op.drop_table('extra')
-    op.add_column('address', sa.Column('street', sa.String(length=50), \
-nullable=True))
+    op.add_column('address', sa.Column('street', sa.String(length=50), nullable=True))
     op.create_unique_constraint('uq_email', 'address', ['email_address'])
     op.add_column('order', sa.Column('user_id', sa.Integer(), nullable=True))
     op.alter_column('order', 'amount',
@@ -109,17 +109,16 @@ nullable=True))
                existing_type=sa.TEXT(),
                server_default='x',
                existing_nullable=True)
-    op.drop_index('pw_idx', table_name='user')
+    op.drop_index(op.f('pw_idx'), table_name='user')
     op.drop_column('user', 'pw')
-    # ### end Alembic commands ###""",
+    # ### end Alembic commands ###""",  # noqa: E501,
         )
 
         eq_(
             template_args["downgrades"],
             """# ### commands auto generated by Alembic - please adjust! ###
-    op.add_column('user', sa.Column('pw', sa.VARCHAR(length=50), \
-nullable=True))
-    op.create_index('pw_idx', 'user', ['pw'], unique=False)
+    op.add_column('user', sa.Column('pw', sa.VARCHAR(length=50), nullable=True))
+    op.create_index(op.f('pw_idx'), 'user', ['pw'], unique=False)
     op.alter_column('user', 'a1',
                existing_type=sa.TEXT(),
                server_default=None,
@@ -142,7 +141,7 @@ nullable=True))
     sa.ForeignKeyConstraint(['uid'], ['user.id'], )
     )
     op.drop_table('item')
-    # ### end Alembic commands ###""",
+    # ### end Alembic commands ###""",  # noqa: E501
         )
 
     def test_render_diffs_batch(self):
@@ -185,7 +184,7 @@ nullable=True))
                existing_type=sa.TEXT(),
                server_default='x',
                existing_nullable=True)
-        batch_op.drop_index('pw_idx')
+        batch_op.drop_index(batch_op.f('pw_idx'))
         batch_op.drop_column('pw')
 
     # ### end Alembic commands ###""",  # noqa,
@@ -196,7 +195,7 @@ nullable=True))
             """# ### commands auto generated by Alembic - please adjust! ###
     with op.batch_alter_table('user', schema=None) as batch_op:
         batch_op.add_column(sa.Column('pw', sa.VARCHAR(length=50), nullable=True))
-        batch_op.create_index('pw_idx', ['pw'], unique=False)
+        batch_op.create_index(batch_op.f('pw_idx'), ['pw'], unique=False)
         batch_op.alter_column('a1',
                existing_type=sa.TEXT(),
                server_default=None,
@@ -350,79 +349,219 @@ class AutogenerateDiffTestWSchema(ModelOne, AutogenTest, TestBase):
 
         eq_(
             template_args["upgrades"],
-            """# ### commands auto generated by Alembic - please adjust! ###
+            f"""# ### commands auto generated by Alembic - please adjust! ###
     op.create_table('item',
     sa.Column('id', sa.Integer(), nullable=False),
     sa.Column('description', sa.String(length=100), nullable=True),
     sa.Column('order_id', sa.Integer(), nullable=True),
     sa.CheckConstraint('len(description) > 5'),
-    sa.ForeignKeyConstraint(['order_id'], ['%(schema)s.order.order_id'], ),
+    sa.ForeignKeyConstraint(['order_id'], ['test_schema.order.order_id'], ),
     sa.PrimaryKeyConstraint('id'),
-    schema='%(schema)s'
+    schema='{self.schema}'
     )
-    op.drop_table('extra', schema='%(schema)s')
-    op.add_column('address', sa.Column('street', sa.String(length=50), \
-nullable=True), schema='%(schema)s')
-    op.create_unique_constraint('uq_email', 'address', ['email_address'], \
-schema='test_schema')
-    op.add_column('order', sa.Column('user_id', sa.Integer(), nullable=True), \
-schema='%(schema)s')
+    op.drop_table('extra', schema='{self.schema}')
+    op.add_column('address', sa.Column('street', sa.String(length=50), nullable=True), schema='{self.schema}')
+    op.create_unique_constraint('uq_email', 'address', ['email_address'], schema='{self.schema}')
+    op.add_column('order', sa.Column('user_id', sa.Integer(), nullable=True), schema='{self.schema}')
     op.alter_column('order', 'amount',
                existing_type=sa.NUMERIC(precision=8, scale=2),
                type_=sa.Numeric(precision=10, scale=2),
                nullable=True,
                existing_server_default=sa.text('0'),
-               schema='%(schema)s')
-    op.create_foreign_key(None, 'order', 'user', ['user_id'], ['id'], \
-source_schema='%(schema)s', referent_schema='%(schema)s')
+               schema='{self.schema}')
+    op.create_foreign_key(None, 'order', 'user', ['user_id'], ['id'], source_schema='{self.schema}', referent_schema='{self.schema}')
     op.alter_column('user', 'name',
                existing_type=sa.VARCHAR(length=50),
                nullable=False,
-               schema='%(schema)s')
+               schema='{self.schema}')
     op.alter_column('user', 'a1',
                existing_type=sa.TEXT(),
                server_default='x',
                existing_nullable=True,
-               schema='%(schema)s')
-    op.drop_index('pw_idx', table_name='user', schema='test_schema')
-    op.drop_column('user', 'pw', schema='%(schema)s')
-    # ### end Alembic commands ###"""
-            % {"schema": self.schema},
+               schema='{self.schema}')
+    op.drop_index(op.f('pw_idx'), table_name='user', schema='{self.schema}')
+    op.drop_column('user', 'pw', schema='{self.schema}')
+    # ### end Alembic commands ###""",  # noqa: E501
         )
 
         eq_(
             template_args["downgrades"],
-            """# ### commands auto generated by Alembic - please adjust! ###
-    op.add_column('user', sa.Column('pw', sa.VARCHAR(length=50), \
-autoincrement=False, nullable=True), schema='%(schema)s')
-    op.create_index('pw_idx', 'user', ['pw'], unique=False, schema='%(schema)s')
+            f"""# ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('user', sa.Column('pw', sa.VARCHAR(length=50), autoincrement=False, nullable=True), schema='{self.schema}')
+    op.create_index(op.f('pw_idx'), 'user', ['pw'], unique=False, schema='{self.schema}')
     op.alter_column('user', 'a1',
                existing_type=sa.TEXT(),
                server_default=None,
                existing_nullable=True,
-               schema='%(schema)s')
+               schema='{self.schema}')
     op.alter_column('user', 'name',
                existing_type=sa.VARCHAR(length=50),
                nullable=True,
-               schema='%(schema)s')
-    op.drop_constraint(None, 'order', schema='%(schema)s', type_='foreignkey')
+               schema='{self.schema}')
+    op.drop_constraint(None, 'order', schema='{self.schema}', type_='foreignkey')
     op.alter_column('order', 'amount',
                existing_type=sa.Numeric(precision=10, scale=2),
                type_=sa.NUMERIC(precision=8, scale=2),
                nullable=False,
                existing_server_default=sa.text('0'),
-               schema='%(schema)s')
-    op.drop_column('order', 'user_id', schema='%(schema)s')
-    op.drop_constraint('uq_email', 'address', schema='test_schema', type_='unique')
-    op.drop_column('address', 'street', schema='%(schema)s')
+               schema='{self.schema}')
+    op.drop_column('order', 'user_id', schema='{self.schema}')
+    op.drop_constraint('uq_email', 'address', schema='{self.schema}', type_='unique')
+    op.drop_column('address', 'street', schema='{self.schema}')
     op.create_table('extra',
     sa.Column('x', sa.CHAR(length=1), autoincrement=False, nullable=True),
     sa.Column('uid', sa.INTEGER(), autoincrement=False, nullable=True),
-    sa.ForeignKeyConstraint(['uid'], ['%(schema)s.user.id'], \
-name='extra_uid_fkey'),
-    schema='%(schema)s'
+    sa.ForeignKeyConstraint(['uid'], ['test_schema.user.id'], name=op.f('extra_uid_fkey')),
+    schema='{self.schema}'
+    )
+    op.drop_table('item', schema='{self.schema}')
+    # ### end Alembic commands ###""",  # noqa
+        )
+
+
+class AutogenerateNamingConvTest(NamingConvModel, AutogenTest, TestBase):
+    __only_on__ = "sqlite"
+
+    def _lines_equal(self, a, b):
+        # compare that a and b have all the same codelines, with the order
+        # not mattering (we are looking just for the right op.f() calls)
+        eq_(set(a.split("\n")), set(b.split("\n")))
+
+    def test_render_diffs_naming_conv(self):
+        template_args = {}
+        autogenerate._render_migration_diffs(self.context, template_args)
+        self._lines_equal(
+            template_args["upgrades"],
+            """# ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('x6',
+    sa.Column('q', sa.Integer(), nullable=False),
+    sa.Column('p', sa.Integer(), nullable=True),
+    sa.Column('r', sa.Integer(), nullable=True),
+    sa.Column('s', sa.Integer(), nullable=True),
+    sa.CheckConstraint('s < 20', name=op.f('userdef_x6_check_s')),
+    sa.CheckConstraint('s > 5', name=op.f('ck_x6_token_x6check1')),
+    sa.ForeignKeyConstraint(['p'], ['x4.q'], name=op.f('fk_x6_p_x4')),
+    sa.PrimaryKeyConstraint('q', name=op.f('pk_x6')),
+    sa.UniqueConstraint('r', name=op.f('uq_x6_token_x6r'))
+    )
+    op.drop_table('unnamed_sqlite')
+    op.drop_table('x5')
+    op.drop_index(op.f('db_x1_index_q'), table_name='x1')
+    op.create_index(op.f('ix_x1_q'), 'x1', ['q'], unique=False)
+    op.drop_constraint(op.f('db_x3_unique_q'), 'x3', type_='unique')
+    op.create_unique_constraint(op.f('uq_x3_token_x3r'), 'x3', ['r'])
+    op.create_unique_constraint(op.f('userdef_x3_unique_s'), 'x3', ['s'])
+    op.create_index('userdef_x4_idx_q', 'x4', ['q'], unique=False)
+    # ### end Alembic commands ###""",  # noqa: E501
+        )
+
+        self._lines_equal(
+            template_args["downgrades"],
+            """# ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index('userdef_x4_idx_q', table_name='x4')
+    op.drop_constraint(op.f('userdef_x3_unique_s'), 'x3', type_='unique')
+    op.drop_constraint(op.f('uq_x3_token_x3r'), 'x3', type_='unique')
+    op.create_unique_constraint(op.f('db_x3_unique_q'), 'x3', ['q'])
+    op.drop_index(op.f('ix_x1_q'), table_name='x1')
+    op.create_index(op.f('db_x1_index_q'), 'x1', ['q'], unique=False)
+    op.create_table('x5',
+    sa.Column('q', sa.INTEGER(), nullable=False),
+    sa.Column('p', sa.INTEGER(), nullable=True),
+    sa.Column('r', sa.INTEGER(), nullable=True),
+    sa.Column('s', sa.INTEGER(), nullable=True),
+    sa.CheckConstraint('s > 5', name=op.f('db_x5_check_s')),
+    sa.ForeignKeyConstraint(['p'], ['x4.q'], name=op.f('db_x5_foreign_q')),
+    sa.PrimaryKeyConstraint('q', name=op.f('db_x5_primary_q')),
+    sa.UniqueConstraint('r', name=op.f('db_x5_unique_r'))
+    )
+    op.create_table('unnamed_sqlite',
+    sa.Column('q', sa.INTEGER(), nullable=False),
+    sa.Column('r', sa.INTEGER(), nullable=True),
+    sa.PrimaryKeyConstraint('q'),
+    sa.UniqueConstraint('r')
+    )
+    op.drop_table('x6')
+    # ### end Alembic commands ###""",  # noqa: E501
+        )
+
+
+class AutogenerateNamingConvWBatchTest(NamingConvModel, AutogenTest, TestBase):
+    __only_on__ = "sqlite"
+    configure_opts = {
+        "conv_all_constraint_names": True,
+        "render_as_batch": True,
+    }
+
+    def _lines_equal(self, a, b):
+        # compare that a and b have all the same codelines, with the order
+        # not mattering (we are looking just for the right op.f() calls)
+        eq_(set(a.split("\n")), set(b.split("\n")))
+
+    def test_render_diffs_naming_conv(self):
+        template_args = {}
+        autogenerate._render_migration_diffs(self.context, template_args)
+        self._lines_equal(
+            template_args["upgrades"],
+            """# ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('x6',
+    sa.Column('q', sa.Integer(), nullable=False),
+    sa.Column('p', sa.Integer(), nullable=True),
+    sa.Column('r', sa.Integer(), nullable=True),
+    sa.Column('s', sa.Integer(), nullable=True),
+    sa.CheckConstraint('s < 20', name=op.f('userdef_x6_check_s')),
+    sa.CheckConstraint('s > 5', name=op.f('ck_x6_token_x6check1')),
+    sa.ForeignKeyConstraint(['p'], ['x4.q'], name=op.f('fk_x6_p_x4')),
+    sa.PrimaryKeyConstraint('q', name=op.f('pk_x6')),
+    sa.UniqueConstraint('r', name=op.f('uq_x6_token_x6r'))
+    )
+    op.drop_table('x5')
+    op.drop_table('unnamed_sqlite')
+    with op.batch_alter_table('x1', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('db_x1_index_q'))
+        batch_op.create_index(batch_op.f('ix_x1_q'), ['q'], unique=False)
+
+    with op.batch_alter_table('x3', schema=None) as batch_op:
+        batch_op.drop_constraint(batch_op.f('db_x3_unique_q'), type_='unique')
+        batch_op.create_unique_constraint(batch_op.f('uq_x3_token_x3r'), ['r'])
+        batch_op.create_unique_constraint(batch_op.f('userdef_x3_unique_s'), ['s'])
+
+    with op.batch_alter_table('x4', schema=None) as batch_op:
+        batch_op.create_index('userdef_x4_idx_q', ['q'], unique=False)
+
+    # ### end Alembic commands ###""",  # noqa: E501
+        )
+
+        self._lines_equal(
+            template_args["downgrades"],
+            """# ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('x4', schema=None) as batch_op:
+        batch_op.drop_index('userdef_x4_idx_q')
+
+    with op.batch_alter_table('x3', schema=None) as batch_op:
+        batch_op.drop_constraint(batch_op.f('userdef_x3_unique_s'), type_='unique')
+        batch_op.drop_constraint(batch_op.f('uq_x3_token_x3r'), type_='unique')
+        batch_op.create_unique_constraint(batch_op.f('db_x3_unique_q'), ['q'])
+
+    with op.batch_alter_table('x1', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_x1_q'))
+        batch_op.create_index(batch_op.f('db_x1_index_q'), ['q'], unique=False)
+
+    op.create_table('unnamed_sqlite',
+    sa.Column('q', sa.INTEGER(), nullable=False),
+    sa.Column('r', sa.INTEGER(), nullable=True),
+    sa.PrimaryKeyConstraint('q'),
+    sa.UniqueConstraint('r')
+    )
+    op.create_table('x5',
+    sa.Column('q', sa.INTEGER(), nullable=False),
+    sa.Column('p', sa.INTEGER(), nullable=True),
+    sa.Column('r', sa.INTEGER(), nullable=True),
+    sa.Column('s', sa.INTEGER(), nullable=True),
+    sa.CheckConstraint('s > 5', name=op.f('db_x5_check_s')),
+    sa.ForeignKeyConstraint(['p'], ['x4.q'], name=op.f('db_x5_foreign_q')),
+    sa.PrimaryKeyConstraint('q', name=op.f('db_x5_primary_q')),
+    sa.UniqueConstraint('r', name=op.f('db_x5_unique_r'))
     )
-    op.drop_table('item', schema='%(schema)s')
-    # ### end Alembic commands ###"""  # noqa
-            % {"schema": self.schema},
+    op.drop_table('x6')
+    # ### end Alembic commands ###""",  # noqa: E501
         )