def _fk_colspec(
fk: ForeignKey,
metadata_schema: Optional[str],
- namespace_metadata: MetaData,
+ namespace_metadata: Optional[MetaData],
) -> str:
"""Implement a 'safe' version of ForeignKey._get_colspec() that
won't fail if the remote table can't be resolved.
# the FK constraint needs to be rendered in terms of the column
# name.
- if table_fullname in namespace_metadata.tables:
+ if (
+ namespace_metadata is not None
+ and table_fullname in namespace_metadata.tables
+ ):
col = namespace_metadata.tables[table_fullname].c.get(colname)
if col is not None:
colname = _ident(col.name) # type: ignore[assignment]
def _render_foreign_key(
constraint: ForeignKeyConstraint,
autogen_context: AutogenContext,
- namespace_metadata: MetaData,
+ namespace_metadata: Optional[MetaData],
) -> Optional[str]:
rendered = _user_defined_render("foreign_key", constraint, autogen_context)
if rendered is not False:
_populate_render_fk_opts(constraint, opts)
- apply_metadata_schema = namespace_metadata.schema
+ apply_metadata_schema = (
+ namespace_metadata.schema if namespace_metadata is not None else None
+ )
return (
"%(prefix)sForeignKeyConstraint([%(cols)s], "
"[%(refcols)s], %(args)s)"
import sqlalchemy as sa
from sqlalchemy import Column
+from sqlalchemy import DateTime
+from sqlalchemy import ForeignKey
from sqlalchemy import inspect
+from sqlalchemy import Integer
from sqlalchemy import MetaData
+from sqlalchemy import String
from sqlalchemy import Table
+from sqlalchemy import UniqueConstraint
from alembic import autogenerate
from alembic import command
eq_(diffs[1][0], "execute")
eq_(diffs[1][1], "STATEMENT")
+ def test_reordering_example_wo_copy(self):
+ """test related to #1692 in that we identified the recipe
+ for rewriting column ordering was using column.copy()"""
+
+ m = MetaData()
+ t1 = Table(
+ "my_table",
+ m,
+ Column("data", String(50)),
+ Column("created_at", DateTime),
+ Column("id", Integer, primary_key=True),
+ Column("updated_at", DateTime),
+ UniqueConstraint("data", name="uq_data"),
+ )
+
+ t2 = Table(
+ "my_other_table",
+ m,
+ Column("data", String(50)),
+ Column("created_at", DateTime),
+ Column("id", Integer, primary_key=True),
+ Column("other_id", ForeignKey("my_table.id")),
+ Column("updated_at", DateTime),
+ UniqueConstraint("data", name="uq_data"),
+ )
+
+ writer = autogenerate.Rewriter()
+
+ @writer.rewrites(ops.CreateTableOp)
+ def order_columns(context, revision, op):
+
+ special_names = {
+ "id": -100,
+ "created_at": 1001,
+ "updated_at": 1002,
+ }
+
+ cols_by_key = [
+ (
+ (
+ special_names.get(col.key, index)
+ if isinstance(col, Column)
+ else 2000
+ ),
+ col, # note no copy()
+ )
+ for index, col in enumerate(op.columns)
+ ]
+
+ columns = [
+ col
+ for idx, col in sorted(cols_by_key, key=lambda entry: entry[0])
+ ]
+ return ops.CreateTableOp(
+ op.table_name, columns, schema=op.schema, **op.kw
+ )
+
+ directives = [
+ ops.MigrationScript(
+ util.rev_id(),
+ ops.UpgradeOps(
+ ops=[
+ ops.CreateTableOp.from_table(t1),
+ ops.CreateTableOp.from_table(t2),
+ ]
+ ),
+ ops.DowngradeOps(ops=[]),
+ )
+ ]
+
+ ctx, rev = mock.Mock(), mock.Mock()
+ writer(ctx, rev, directives)
+ eq_(
+ autogenerate.render_python_code(directives[0].upgrade_ops),
+ "# ### commands auto generated by Alembic - please adjust! ###\n"
+ " op.create_table('my_table',\n"
+ " sa.Column('id', sa.Integer(), nullable=False),\n"
+ " sa.Column('data', sa.String(length=50), nullable=True),\n"
+ " sa.Column('created_at', sa.DateTime(), nullable=True),\n"
+ " sa.Column('updated_at', sa.DateTime(), nullable=True),\n"
+ " sa.PrimaryKeyConstraint('id'),\n"
+ " sa.UniqueConstraint('data', name='uq_data')\n"
+ " )\n"
+ " op.create_table('my_other_table',\n"
+ " sa.Column('id', sa.Integer(), nullable=False),\n"
+ " sa.Column('data', sa.String(length=50), nullable=True),\n"
+ " sa.Column('other_id', sa.Integer(), nullable=True),\n"
+ " sa.Column('created_at', sa.DateTime(), nullable=True),\n"
+ " sa.Column('updated_at', sa.DateTime(), nullable=True),\n"
+ " sa.ForeignKeyConstraint(['other_id'], ['my_table.id'], ),\n"
+ " sa.PrimaryKeyConstraint('id'),\n"
+ " sa.UniqueConstraint('data', name='uq_data')\n"
+ " )\n"
+ " # ### end Alembic commands ###",
+ )
+
class MultiDirRevisionCommandTest(TestBase):
def setUp(self):