python - Alembic 迁移无法正确生成 UUID

Question

事实上确实如此，但我肯定做错了什么。

我有一个数据库，我在表上创建了一个 FK，在未来的迁移中，它被删除了。

所以表的create语句为：

CREATE TABLE this_table (
    pk INTEGER NOT NULL,
    some_other_table_pk INTEGER,
    CONSTRAINT "fk_436d3a57-0ae2-5668-937d-8cbc82d15172" FOREIGN KEY(some_other_table_pk) REFERENCES some_other_table (pk)

然后我放弃约束，alembic 自动生成以下批处理操作：

def upgrade():
    with op.batch_alter_table('this_table', schema=None) as batch_op:
        # how does it know here which fk to drop? there are more than one???
        batch_op.drop_constraint(None, type_='foreignkey')
        batch_op.drop_column('some_other_table_pk')

def downgrade():
    with op.batch_alter_table('this_table', schema=None) as batch_op:
        batch_op.add_column(sa.Column('some_other_table_pk', sa.INTEGER(), nullable=True))
        batch_op.create_foreign_key(None, 'some_other_table_pk', ['some_other_table'], ['pk'])

这导致升级头错误：

KeyError: 'fk_ad156c6d-7a24-5d8e-868e-2fb00c2cb1c9'

很明显它没有重新生成相同的密钥？！？该密钥与此表上的任何 fk 都不匹配...但是它应该如何知道要生成哪个密钥？drop_constraint 没有得到关于要删除哪个 fk 的任何信息？

我的元数据约定：

def fk_guid(constraint, table):
    str_tokens = [
        table.name,
    ] + [
        element.parent.name for element in constraint.elements
    ] + [
        element.target_fullname for element in constraint.elements
    ]
    guid = uuid.uuid5(uuid.NAMESPACE_OID, "_".join(str_tokens))
    return str(guid)

convention = {
  "fk_guid": fk_guid,
  "ix": "ix_%(column_0_label)s",
  "uq": "uq_%(table_name)s_%(column_0_name)s",
  "ck": "ck_%(table_name)s_%(column_0_name)s",
  "fk": "fk_%(fk_guid)s",
  "pk": "pk_%(table_name)s",
}

metadata = MetaData(naming_convention=convention)

Base = declarative_base(metadata=metadata)

在 env.py 中：

# ...
target_metadata = models.Base.metadata
# ...
def run_migrations_offline():
    url = getUrl()
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        render_as_batch=True,
        dialect_opts={"paramstyle": "named"},
        compare_type=True
    )

    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online():
    alembic_config = config.get_section(config.config_ini_section)

    connectable = create_engine(getUrl())

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            render_as_batch=True,
            dialect_opts={"paramstyle": "named"},
            compare_type=True
        )

        with context.begin_transaction():
            context.run_migrations()

数据库是一个 sqlite3 文件。

编辑开始

所以我尝试手动传入要删除的约束名称：

def upgrade():
    with op.batch_alter_table('this_table', schema=None) as batch_op:
        batch_op.drop_constraint('fk_436d3a57-0ae2-5668-937d-8cbc82d15172', type_='foreignkey')
        batch_op.drop_column('some_other_table_pk')

现在它导致了一个关键错误，但关键应该已经存在于数据库中？！？

KeyError: 'fk_436d3a57-0ae2-5668-937d-8cbc82d15172'

编辑结束

编辑开始2

好的，所以我在沙箱中玩了一点，似乎它是用户定义的令牌密钥，如果它是 drop_constraint 则无法升级：

以下情况下的 fk_guid 仅在创建约束时调用

convention = {
  "fk_guid": some_function,
  "ix": "ix_%(column_0_label)s",
  "uq": "uq_%(table_name)s_%(column_0_name)s",
  "ck": "ck_%(table_name)s_%(column_0_name)s",
  "fk": "fk_%(fk_guid)s",
#  "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
  "pk": "pk_%(table_name)s",
}

并给出以下用于创建和删除的迁移代码：

sa.ForeignKeyConstraint(['table_b'], ['table_b.id'], name=op.f('fk_9d68bd30-b17a-565a-aee7-7bd21e90672e')),

# But for dropping, it's not generated:
batch_op.drop_constraint(None, type_='foreignkey')

但奇怪的some_function是，被调用alembic upgrade head但与创建调用不同的约束：

def fk_guid(constraint, table):
    str_tokens = [
        table.name,
    ] + [
        element.parent.name for element in constraint.elements
    ] + [
        element.target_fullname for element in constraint.elements
    ]

    print(f'string_tokens: {str_tokens}')

    guid = uuid.uuid5(uuid.NAMESPACE_OID, "_".join(str_tokens))
    return str(guid)

def some_function(*params):
    print(f'params: {params}')
    return fk_guid(*params)

输出：

string_tokens: ['table_a']

将外键的命名约定更改为标准

convention = {
  "fk_guid": some_function,
  "ix": "ix_%(column_0_label)s",
  "uq": "uq_%(table_name)s_%(column_0_name)s",
  "ck": "ck_%(table_name)s_%(column_0_name)s",
#  "fk": "fk_%(fk_guid)s",
  "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
  "pk": "pk_%(table_name)s",
}

根据需要进行创建和删除：

# creation:
sa.ForeignKeyConstraint(['table_b'], ['table_b.id'], name=op.f('fk_table_a_table_b_table_b')),

# dropping:
batch_op.drop_constraint('fk_table_a_table_b_table_b', type_='foreignkey')

编辑结束2

编辑开始3

github上有一个线程：https ://github.com/sqlalchemy/alembic/issues/713

编辑结束3

标签： pythonsqlalchemymigrationuuidalembic