使用Alembic从现有数据库生成初始迁移时表被误判为已删除的问题

问题描述 #

需要从包含33张表、键、序列的现有数据库创建Alembic迁移，但无SQLAlchemy元数据和ORM类。使用automap_base生成映射后，Alembic将所有表标记为已删除，生成的迁移文件upgrade为删表逻辑，downgrade为建表逻辑，手动调换后创建新迁移时提示数据库未处于最新状态。

用到的映射代码（alchemy_mapping.py） #

from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import Session
from sqlalchemy import create_engine

class MigrationMeta:

    Base = automap_base()

    # engine, suppose it has two tables 'user' and 'address' set up
    engine = create_engine("postgresql://user:pass@host/db")
    def __init__(self):
        self.Base.prepare(autoload_with=self.engine, reflect=True)
        self.Base.metadata.create_all(self.engine)

env.py中的配置 #

from api.alchemy_mapping import MigrationMeta 
target_metadata = MigrationMeta.Base.metadata

自动生成后的错误日志 #

...
INFO  [alembic.autogenerate.compare] Detected removed table 'user_setting'
INFO  [alembic.autogenerate.compare] Detected removed table 'payment_accounts'
INFO  [alembic.autogenerate.compare] Detected removed table 'user_tier'
INFO  [alembic.autogenerate.compare] Detected removed table 'mobule_game'
...

生成的迁移文件片段 #

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('bonus_settings')
    op.drop_table('contractor')
    op.drop_index('tokens_token_index', table_name='tokens')
    op.drop_index('tokens_token_token_ip_index', table_name='tokens')
    op.drop_index('tokens_user_id_index', table_name='tokens')
    op.drop_table('tokens')
    op.drop_table('game')
...

def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('method_access_right',
    sa.Column('id', sa.INTEGER(), server_default=sa.text("nextval('method_access_right_id_seq'::regclass)"), autoincrement=True, nullable=False),
    sa.Column('method', sa.VARCHAR(length=32), autoincrement=False, nullable=False),
    sa.Column('project_type', sa.VARCHAR(length=32), autoincrement=False, nullable=False),
    sa.Column('user_role', sa.VARCHAR(length=128), autoincrement=False, nullable=True),
    sa.Column('admin_system', sa.VARCHAR(length=128), autoincrement=False, nullable=False)
    )
...

问题原因 #

核心错误是元数据未被正确初始化：

• 在env.py中直接引用MigrationMeta.Base.metadata，但未实例化MigrationMeta类，导致Base.prepare(autoload_with=self.engine, reflect=True)从未执行，target_metadata是空的。
• Alembic对比空元数据和现有数据库时，会误认为所有表都是多余的，因此生成upgrade删表、downgrade建表的反向逻辑。
• 手动调换upgrade和downgrade后，Alembic的版本跟踪逻辑与实际数据库状态不匹配，导致后续提示数据库未处于最新状态。

解决步骤 #

1. 修正映射代码（alchemy_mapping.py） #

移除不必要的类封装，直接生成并导出反射后的元数据，避免实例化问题：

from sqlalchemy.ext.automap import automap_base
from sqlalchemy import create_engine

# 初始化数据库连接引擎
engine = create_engine("postgresql://user:pass@host/db")
# 创建automap基础类
Base = automap_base()
# 执行反射，加载现有数据库结构到元数据
Base.prepare(autoload_with=engine, reflect=True)

# 直接导出元数据供Alembic使用
target_metadata = Base.metadata

2. 更新env.py配置 #

直接导入正确的元数据：

from api.alchemy_mapping import target_metadata

3. 生成并标记初始迁移 #

• 运行命令生成初始迁移文件：

  alembic revision --autogenerate -m "initial schema"
  

  此时生成的迁移文件中，upgrade包含创建所有现有表的逻辑，downgrade包含删除逻辑。
• 不要执行alembic upgrade head（表已存在，执行会报错），而是运行命令将当前数据库标记为已应用该迁移：

  alembic stamp head
  

4. 后续迁移流程 #

后续修改数据库结构（或更新ORM模型）后，直接运行alembic revision --autogenerate生成新迁移，再执行alembic upgrade head应用变更即可。

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