如何用Alembic为同一库下不同schema的微服务做独立迁移管理

该需求完全可以实现，无需额外引入工具，仅通过Alembic原生配置即可完成各服务迁移的完全隔离。

1. 初始化各服务的Alembic环境 #

首先保证每个服务单独管理自己的依赖，在requirements.txt中添加alembic和对应数据库驱动（如PostgreSQL用psycopg2-binary，MySQL用pymysql）。
分别进入两个服务目录，执行初始化命令：

# 进入service1目录执行
cd team/service1
alembic init alembic

# 进入service2目录执行
cd ../service2
alembic init alembic

执行完成后即可得到你期望的目录结构。

2. 配置数据库连接 #

分别修改两个服务下的alembic.ini文件，找到sqlalchemy.url配置项，填入同一个数据库的连接串即可，无需在连接串中指定schema：

# 示例（PostgreSQL），两个服务的该配置一致即可
sqlalchemy.url = postgresql://username:password@数据库地址:端口/数据库名

3. 配置Schema隔离规则 #

分别修改两个服务下的alembic/env.py文件，核心是指定当前服务对应的专属schema，同时将Alembic的版本记录表也放到对应schema下，避免跨服务冲突。

3.1 定义目标Schema #

在env.py的导入语句后添加常量定义：

# service1的env.py加这行
TARGET_SCHEMA = "service1"

# service2的env.py加这行
TARGET_SCHEMA = "service2"

3.2 修改离线迁移配置 #

找到run_migrations_offline函数，在context.configure参数中新增schema相关配置，修改后示例：

def run_migrations_offline():
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata, # 这里保持你自己的模型元数据配置即可
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
        # 新增以下两行配置
        schema_translate_map={None: TARGET_SCHEMA},
        version_table_schema=TARGET_SCHEMA,
    )
    with context.begin_transaction():
        # 可选：自动创建不存在的schema
        context.execute(f"CREATE SCHEMA IF NOT EXISTS {TARGET_SCHEMA}")
        context.run_migrations()

3.3 修改在线迁移配置 #

找到run_migrations_online函数，同样在context.configure参数中新增schema相关配置，修改后示例：

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

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata, # 这里保持你自己的模型元数据配置即可
            # 新增以下两行配置
            schema_translate_map={None: TARGET_SCHEMA},
            version_table_schema=TARGET_SCHEMA,
        )

        with context.begin_transaction():
            # 可选：自动创建不存在的schema
            connection.execute(text(f"CREATE SCHEMA IF NOT EXISTS {TARGET_SCHEMA}"))
            connection.commit()
            context.run_migrations()

注：如果使用上面的自动创建schema逻辑，需要提前在env.py中导入from sqlalchemy import text。

完全符合你期望的操作逻辑：

• 进入team/service1目录，执行alembic revision --autogenerate -m "你的迁移说明"生成迁移脚本，所有生成的表操作都会默认指向service1 schema
• 执行alembic upgrade head，仅会修改service1 schema下的表，版本记录也存在service1 schema下，完全不影响其他服务
• 进入team/service2目录执行上述相同操作，仅会操作service2 schema下的资源

• 确保数据库账号拥有两个schema的读写权限
• 如果你的模型代码中手动指定了schema，需要和当前服务的TARGET_SCHEMA保持一致，避免出现跨schema操作
• 两个服务的Alembic环境完全独立，不要跨目录执行迁移命令

内容的提问来源于stack exchange，提问作者Be Chiller Too