如何使用Flask-Migrate(Alembic)仅升级或迁移单个数据库Schema？

方法1：直接使用Alembic的--schema参数 #

虽然Flask-Migrate封装了Alembic，但你可以绕过Flask-Script，直接调用Alembic命令指定目标Schema：

• 生成仅针对目标Schema的迁移脚本：

  alembic revision --autogenerate --schema=your_target_schema
  

• 升级指定Schema：

  alembic upgrade head --schema=your_target_schema
  

注意：确保你的SQLAlchemy模型类已通过__table_args__绑定到目标Schema，示例：

class YourModel(db.Model):
    __tablename__ = 'your_table'
    __table_args__ = {'schema': 'your_target_schema'}
    # 字段定义...

方法2：修改Alembic的env.py过滤Schema #

通过修改迁移环境配置文件env.py，让Flask-Migrate仅处理指定Schema的变更：

1. 在env.py中添加环境变量读取和对象过滤逻辑：

import os

def include_object(object, name, type_, reflected, compare_to):
    # 读取环境变量指定的目标Schema
    target_schema = os.environ.get('TARGET_SCHEMA')
    if not target_schema:
        return True  # 未指定时处理所有Schema
    # 只保留目标Schema下的表
    if type_ == 'table' and object.schema != target_schema:
        return False
    return True

2. 在run_migrations_online函数的context.configure中添加过滤配置：

context.configure(
    connection=connection,
    target_metadata=target_metadata,
    include_schemas=True,  # 开启多Schema支持
    include_object=include_object,  # 应用过滤逻辑
    # 其他原有配置...
)

3. 执行迁移时指定目标Schema：

# Linux/macOS
export TARGET_SCHEMA=your_target_schema
# Windows
set TARGET_SCHEMA=your_target_schema

# 生成迁移
flask db migrate -m "仅更新your_target_schema"
# 执行升级
flask db upgrade

方法3：自定义Flask-Script命令封装单Schema操作 #

如果习惯用Flask-Script的命令行方式，可以自定义命令封装单Schema的迁移和升级：

from flask_script import Command, Manager
from alembic.config import Config
from alembic import command

# 假设你的Flask app和migrate已初始化
manager = Manager(app)

class SchemaMigrateCommand(Command):
    def __init__(self, schema_name, message):
        self.schema_name = schema_name
        self.message = message

    def run(self):
        alembic_cfg = Config("alembic.ini")
        command.revision(alembic_cfg, autogenerate=True, message=self.message, schema=self.schema_name)

class SchemaUpgradeCommand(Command):
    def __init__(self, schema_name):
        self.schema_name = schema_name

    def run(self):
        alembic_cfg = Config("alembic.ini")
        command.upgrade(alembic_cfg, "head", schema=self.schema_name)

# 注册命令
manager.add_command('db_schema_migrate', SchemaMigrateCommand('your_target_schema', '迁移备注'))
manager.add_command('db_schema_upgrade', SchemaUpgradeCommand('your_target_schema'))

然后直接执行自定义命令：

python manage.py db_schema_migrate
python manage.py db_schema_upgrade

关键提醒：无论用哪种方法，生成迁移脚本后一定要手动检查内容，确认仅包含目标Schema的变更，避免误操作其他Schema的表结构。

内容的提问来源于stack exchange，提问作者Phaneendhar Reddy