如何用Alembic为PostgreSQL多租户Schema自动生成迁移脚本？

问题原因分析 #

1. 模型硬编码Schema：User模型中直接指定schema="test"，导致Alembic自动生成迁移脚本时，默认仅为test schema创建表。
2. 迁移脚本绑定固定Schema：生成的脚本中op.create_table明确指定schema='test'，即便env.py配置了schema_translate_map，也无法替换脚本中硬编码的Schema参数。
3. 版本表位置错误：version_table_schema=target_metadata.schema会把所有租户的迁移版本表集中放在test schema，无法实现租户独立的迁移状态跟踪。

解决方案步骤 #

1. 调整模型Schema配置 #

移除模型中硬编码的Schema，让表不绑定到固定Schema；若需开发环境使用默认Schema，可通过环境变量动态配置：

# models.py
import os
# 开发环境默认用public，生产环境通过环境变量指定
DEFAULT_SCHEMA = os.getenv("DB_SCHEMA", "public")

class User(Base):
    __tablename__ = 'users'
    __table_args__ = ({"schema": DEFAULT_SCHEMA})

    id = Column(Integer, primary_key=True)
    name = Column(String(80), unique=True, nullable=False)

    def __repr__(self):
        return '<User %r>' % self.name

2. 修改Alembic的env.py配置 #

调整迁移逻辑，确保每个租户Schema存在，并动态替换Schema、独立跟踪迁移版本：

# env.py
from alembic import context
from sqlalchemy import engine_from_config, pool
import app  # 导入你的应用模块，包含Base模型

target_metadata = app.Base.metadata

def run_migrations_online() -> None:
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        all_tenants = ["tenant_1", "tenant_2"]
        
        for tenant_schema in all_tenants:
            # 确保租户Schema已创建，不存在则自动创建
            connection.execute(f"CREATE SCHEMA IF NOT EXISTS {tenant_schema}")
            connection.commit()

            # 设置Schema映射：将模型默认Schema替换为当前租户Schema
            conn = connection.execution_options(
                schema_translate_map={DEFAULT_SCHEMA: tenant_schema}
            )

            # 配置迁移上下文：每个租户独立存储迁移版本表
            context.configure(
                connection=conn,
                target_metadata=target_metadata,
                include_schemas=True,
                version_table_schema=tenant_schema,
                # 过滤无关Schema对比，避免干扰迁移逻辑
                include_object=lambda obj, name, type_, reflected, compare_to: 
                    type_ != "schema" or name == tenant_schema
            )

            with context.begin_transaction():
                # 设置数据库搜索路径优先使用当前租户Schema
                context.execute(f"SET search_path TO {tenant_schema}, public")
                context.run_migrations()

if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

3. 重新生成并运行迁移 #

1. 删除旧迁移脚本（避免冲突），执行自动生成命令：

alembic revision --autogenerate -m "create users table for tenants"

此时生成的迁移脚本不会指定固定Schema：

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('users',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('name', sa.String(length=80), nullable=False),
    sa.PrimaryKeyConstraint('id'),
    sa.UniqueConstraint('name')
    )
    # ### end Alembic commands ###

2. 运行迁移脚本，为所有租户创建表：

alembic upgrade head

验证结果 #

登录PostgreSQL后可查看：

• tenant_1.users和tenant_2.users表已成功创建
• 每个租户Schema下都有独立的alembic_version表，各自跟踪迁移状态

内容的提问来源于stack exchange，提问作者Manisha Bayya