python - SqlAlchemy alembic 迁移文件不使用 env.py 中的连接设置？

Question

我有一个适用于名为 tenant_schema 的通用架构的迁移。在 env.py 的 run_migrations_online 函数中，我为 tenant_schema 设置了一个 schema_translate_map。

我希望 sqlalchemy 将此迁移操作转换为在所需模式上运行，但它似乎试图使用模式 tenant_schema 运行 sql 查询。

有什么办法可以解决吗？

示例:

迁移文件中的升级函数:

2018-09-05_17-28_247f3546088f_add_foo_column.py

def upgrade():
    op.add_column('derived_table', sa.Column('foo', sa.BigInteger(), nullable=True), 
                  schema='tenant_schema')

run_migrations_online 函数:

env.py

schema = 'other_name'  # normally i get the name as an argument from alembic
def run_migrations_online():
    connectable = create_engine(get_url(), echo=True)

    with connectable.connect() as connection:
        # setting up the translation map
        conn = connection.execution_options(schema_translate_map={'tenant_schema': schema})
        context.configure(
            connection=conn,
            target_metadata=target_metadata,
            include_schemas=True,
            version_table_schema=schema,
            include_object=include_object,
        )
        with context.begin_transaction():
            context.run_migrations()

异常(完整的回溯太长而且信息量不大):

sqlalchemy.exc.ProgrammingError: (psycopg2.ProgrammingError) schema "tenant_schema" does not exist
 [SQL: 'ALTER TABLE tenant_schema.derived_table ADD COLUMN foo BIGINT'] 
(Background on this error at: http://sqlalche.me/e/f405)

如您所见，它尝试执行 ALTER TABLE tenant_schema.derived_table 而不是所需的 ALTER TABLE other_name.derived_table

Answer 1

问题

来自SQLAlchemy docs关于 schema_translate_map(强调我的):

The feature takes effect only in those cases where the name of the schema is derived directly from that of a Table or Sequence; it does not impact methods where a string schema name is passed directly

由于所有模式都直接在 alembic 迁移操作中传递，因此不考虑 schema_translate_map。

解决方案

您可能需要的是:

1. 使用 alembic 钩子(Hook)调整模式添加到迁移的方式，使其不是文字字符串，而是在某些全局上下文中的查找(例如渲染 os.environ['TENANT_SCHEMA'] 而不是文字字符串 'tenant_schema')。

   Hook 的正确位置可能是覆盖渲染函数，请参阅 docs 中的示例.不幸的是，我无法为此显示任何代码，因为我自己还没有尝试过。

   或者，您可以尝试注册您的自定义比较器，该比较器将在 alembic 之后运行，实际上不会比较任何内容，但会将 alembic 生成的操作中的 schema 属性替换为自定义字符串子类:

   from alembic.autogenerate import comparators
   
   class FakeSchema(str):
       def __repr__(self):
           return "os.environ['TENANT_SCHEMA']"
   
   @comparators.dispatch_for('schema')
   def tweak_schema(autogen_context, upgrade_ops, schemas):
       for op in upgrade_ops.ops:
           if getattr(op, 'schema', None) == 'tenant_schema':
               op.schema = FakeSchema(op.schema)
               autogen_context.imports.add('import os')  # for os.environ
   

   您可以阅读有关比较器函数的信息 in alembic docs .

2. 将该全局上下文中的架构名称设置为运行迁移时所需的值(在此示例中，将 TENANT_SCHEMA 环境变量传递给 alembic 或将其添加到 os. environ 在你的 env.py 中)。