如何在SQLAlchemy与Alembic中使用多PostgreSQL Schema？

核心需求 #

PostgreSQL数据库包含dev和prod两个Schema，需通过Alembic将SQLAlchemy ORM变更先部署到dev测试，再同步到prod，但现有流程存在多个适配问题。

遇到的问题 #

• 单Schema场景下修改ORM的SCHEMA值，每次迁移会删除所有约束后重新添加，迁移脚本冗长难以阅读。
• 修改ORM的SCHEMA后执行alembic upgrade head，第二个Schema能创建表但未生成alembic_versions表，后续同步第一个Schema的变更时提示“数据库未更新”错误。
• 尝试官方多Schema方案后，移除表定义中的SCHEMA设置，出现“找不到Table None.MyTable”错误，无法创建表。
• 尝试社区方案后，Alembic误修改了不应触碰的Schema。

现有代码片段 #

env.py #

from logging.config import fileConfig

from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context
import sys
sys.path.append('..\\model')
from model.model import metadata, SCHEMA

# Alembic配置对象
config = context.config

# 配置日志
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# 关联ORM的MetaData
target_metadata = metadata

def run_migrations_offline() -> None:
    """离线模式运行迁移"""
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

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

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
        future=True
    )

    current_tenant = context.get_x_argument(as_dictionary=True).get("tenant")
    with connectable.connect() as connection:
        connection.dialect.default_schema_name = current_tenant

        context.configure(
            connection=connection,
            target_metadata=target_metadata,
        )

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

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

ORM类定义 #

SCHEMA = 'my_scheme'
metadata = MetaData(schema=SCHEMA)
Base = declarative_base(metadata=metadata)

class CertScheme(Base):
    __table_args__ = {'schema': SCHEMA}
    __tablename__ = 'CertScheme'

    id = Column(BigInteger, primary_key=True)
    name = Column(String(20), nullable=False)
    company = Column(String(70))
    company_url = Column(String(100))
    description = Column(Text)

解决方案 #

一、修正ORM与Alembic的Schema配置 #

1. 移除ORM中的硬编码Schema
   取消MetaData和表定义中的固定Schema设置，改为通过Alembic动态指定：

   metadata = MetaData()
   Base = declarative_base(metadata=metadata)
   
   class CertScheme(Base):
       __tablename__ = 'CertScheme'
       # 移除__table_args__中的schema配置
       id = Column(BigInteger, primary_key=True)
       name = Column(String(20), nullable=False)
       company = Column(String(70))
       company_url = Column(String(100))
       description = Column(Text)
   

2. 重构env.py适配多Schema
   修改在线迁移逻辑，支持通过命令行参数指定目标Schema，确保每个Schema维护独立的alembic_versions表：

   def run_migrations_online():
       connectable = engine_from_config(
           config.get_section(config.config_ini_section),
           prefix="sqlalchemy.",
           poolclass=pool.NullPool,
           future=True
       )
   
       current_schema = context.get_x_argument(as_dictionary=True).get("schema")
       if not current_schema:
           raise ValueError("必须通过--x-schema指定目标Schema，示例：alembic revision --autogenerate -m 'initial' --x-schema=dev")
   
       with connectable.connect() as connection:
           # 设置当前连接的搜索路径到目标Schema
           connection.execute(f"SET search_path TO {current_schema}")
           connection.dialect.default_schema_name = current_schema
   
           # 确保目标Schema存在，不存在则创建
           connection.execute(f"CREATE SCHEMA IF NOT EXISTS {current_schema}")
           connection.commit()
   
           context.configure(
               connection=connection,
               target_metadata=target_metadata,
               version_table_schema=current_schema,  # 版本表存放在目标Schema内
               compare_type=True,
               compare_server_defaults=True,
               render_as_batch=True,  # 启用批量渲染，减少约束删除重建
               include_object=lambda obj, name, type_, reflected, compare_to: (
                   type_ != "table" or obj.schema == current_schema
               )  # 仅处理目标Schema的对象
           )
   
           with context.begin_transaction():
               context.run_migrations()
   

二、解决约束重复删除重建问题 #

启用render_as_batch=True是核心，该配置让Alembic采用批量DDL操作，避免删除所有约束后重建，直接修改列或约束。配合include_object过滤，确保Alembic只关注当前指定Schema的对象，不会误判表结构变化。

三、标准迁移流程 #

1. 生成dev Schema的迁移脚本

   alembic revision --autogenerate -m "初始化表结构" --x-schema=dev
   

2. 执行dev Schema迁移

   alembic upgrade head --x-schema=dev
   

3. 测试通过后生成prod Schema迁移脚本

   alembic revision --autogenerate -m "同步到生产环境" --x-schema=prod
   

4. 执行prod Schema迁移

   alembic upgrade head --x-schema=prod
   

四、修复历史错误状态 #

若某Schema缺失alembic_versions表，可手动标记当前版本：

# 标记dev Schema为最新版本
alembic stamp head --x-schema=dev
# 标记prod Schema为最新版本
alembic stamp head --x-schema=prod

内容的提问来源于stack exchange，提问作者Florian Hilprecht