Alembic autogenerate无法识别PostgreSQL自定义schema变更的配置问询
- PostgreSQL(含自定义schema)
- Sqlalchemy == 2.0.15
- Alembic == 1.11.1
执行alembic revision --autogenerate -m ".."时,生成的DDL会包含所有数据库对象,仿佛指定的events schema不存在。项目中已正确创建绑定events schema的元数据:
self.sa_metadata: MetaData = MetaData(naming_convention=POSTGRES_INDEXES_NAMING_CONVENTION, schema=self.PG_SCHEMA)
并且已将target_metadata指向该元数据,打印验证schema名称正确。
尝试修改env.py添加include_schemas=True和schema切换语句后,迁移脚本能生成events schema下的正确创建/修改语句,但同时生成了大量不属于目标schema的drop命令(已注释),示例迁移代码如下:
def upgrade() -> None: # ### commands auto generated by Alembic - please adjust! ### op.create_table('access_rights', sa.Column('id', sa.Integer(), autoincrement=True, nullable=False), sa.Column('read', sa.BOOLEAN(), nullable=False), sa.Column('write', sa.BOOLEAN(), nullable=False), sa.Column('change', sa.BOOLEAN(), nullable=False), sa.PrimaryKeyConstraint('id', name=op.f('pk__access_rights')), schema='events' ) op.create_index(op.f('ix__access_rights__id'), 'access_rights', ['id'], unique=False, schema='events') # op.drop_table('alembic_version') # op.drop_index('ix__user_privileges__id', table_name='user_privileges') # op.drop_table('user_privileges') # op.drop_index('ix__event__id', table_name='event') # op.drop_table('event') # op.drop_index('ix__role__id', table_name='role') # op.drop_table('role') # op.drop_index('ix__event_access_rights__id', table_name='event_access_rights') # op.drop_table('event_access_rights') # op.drop_index('ix__user__email', table_name='user') # op.drop_table('user') op.add_column('event_access_rights', sa.Column('rights_id', sa.Integer(), nullable=True), schema='events') op.create_foreign_key(op.f('fk__event_access_rights__rights_id__event'), 'event_access_rights', 'event', ['rights_id'], ['id'], source_schema='events', referent_schema='events') # ### end Alembic commands ###
多余drop命令的根源是include_schemas=True让Alembic扫描了数据库中所有schema的对象,而target_metadata只包含events schema的对象,Alembic误判其他schema(如public)的对象为冗余。按以下步骤调整配置:
移除
include_schemas=True
该参数会强制Alembic遍历所有schema,仅保留target_metadata绑定的schema即可,无需开启全局扫描。配置搜索路径
保留SET search_path TO events语句,确保Alembic连接后默认使用目标schema,避免跨schema的对象识别混乱:context.execute("SET search_path TO events")可选:精确过滤schema(若需保留
include_schemas)
如果必须启用include_schemas=True,添加include_name回调函数,仅允许eventsschema被扫描:def include_name(name, type_, parent_names): if type_ == "schema": return name == target_metadata.schema return True然后在
context.configure中加入该参数:context.configure( connection=connection, target_metadata=target_metadata, version_table_schema=target_metadata.schema, include_schemas=True, include_name=include_name )验证元数据完整性
确认target_metadata下所有表均正确绑定eventsschema,无表误关联到其他schema。
修改后的do_run_migrations函数示例:
def do_run_migrations(connection: Connection) -> None: context.configure( connection=connection, target_metadata=target_metadata, version_table_schema=target_metadata.schema ) with context.begin_transaction(): context.execute("SET search_path TO events") context.run_migrations()
内容的提问来源于stack exchange,提问作者nagibator_archivator

