Alembic自动生成迁移未创建自定义Schema问题排查

通过Base类的@declared_attr装饰器动态生成数据库Schema名称，已在Alembic的env.py中配置include_schemas=True，但自动生成的迁移文件仅在表定义中指定了Schema（如STG_UNION_MEMBER），未包含Schema的创建语句。

1. 手动将Schema加入SQLAlchemy元数据 #

动态生成的Schema不会自动注册到Base的元数据中，需要遍历所有实体类提取Schema名，创建对应的Schema对象并添加到元数据。修改env.py，在target_metadata = Base.metadata之后添加以下代码：

from sqlalchemy.schema import SchemaItem

# 收集所有实体类的Schema名称
schemas = set()
for cls in Base.__subclasses__():
    table_args = getattr(cls, '__table_args__', {})
    if isinstance(table_args, dict) and 'schema' in table_args:
        schemas.add(table_args['schema'])

# 将Schema注册到元数据
for schema in schemas:
    if schema not in target_metadata.schema_names:
        target_metadata.add_entry(SchemaItem(schema))

2. 配置Alembic的Schema检测钩子 #

在env.py的context.configure中添加include_object函数，确保Alembic检测并处理Schema对象：

def include_object(obj, name, type_, reflected, compare_to):
    # 允许检测Schema类型对象
    if type_ == "schema":
        return True
    return True

# 在run_migrations_online和run_migrations_offline的context.configure中加入：
context.configure(
    # 原有参数...
    include_object=include_object,
    include_schemas=True,
)

3. 临时方案：手动修改迁移文件 #

若上述配置未立即生效，可在生成的迁移文件中手动添加Schema创建/删除语句：

def upgrade():
    # 先创建Schema（IF NOT EXISTS避免重复创建）
    op.execute("CREATE SCHEMA IF NOT EXISTS STG_UNION_MEMBER")
    # 原有创建表代码...
    op.create_table('union_member', schema='STG_UNION_MEMBER', ...)

def downgrade():
    # 先删除表，再删除Schema
    op.drop_table('union_member', schema='STG_UNION_MEMBER')
    op.execute("DROP SCHEMA IF EXISTS STG_UNION_MEMBER")

原因说明 #

Alembic的include_schemas=True仅表示迁移会处理Schema下的表，但如果SQLAlchemy元数据中没有Schema本身的定义，Alembic无法检测到需要创建Schema的操作。动态生成的Schema不会自动加入元数据，因此需要手动注册。

内容的提问来源于stack exchange，提问作者Станислав Рославцев