Alembic无法识别多模型及FastAPI异步迁移报错求助

问题概述 #

使用SQLAlchemy[asyncio]、FastAPI、Alembic、asyncmy连接MySQL时，遇到两个核心问题：

• 执行alembic revision --autogenerate生成的迁移脚本中，upgrade()和downgrade()函数为空，未生成表结构变更代码
• 使用默认同步run_migrations_online函数时，抛出sqlalchemy.exc.MissingGreenlet错误，无法正常执行迁移

解决方案 #

1. 修复异步迁移运行错误 #

错误根源是在同步上下文直接调用异步函数，需修改alembic/env.py，用asyncio.run包裹异步迁移逻辑，同时调整连接处理方式：

修改env.py的执行逻辑

替换文件末尾的代码块：

# 替换原有的最后两行
if context.is_offline_mode():
    run_migrations_offline()
else:
    import asyncio
    asyncio.run(run_migrations_online())

更新run_migrations_online函数

确保异步连接和迁移逻辑适配SQLAlchemy 2.0+异步API：

async def run_migrations_online() -> None:
    connectable = AsyncEngine(
        engine_from_config(
            config.get_section(config.config_ini_section, {}),
            prefix="sqlalchemy.",
            poolclass=pool.NullPool,
            future=True,  # 启用SQLAlchemy 2.0+特性
        )
    )

    async with connectable.connect() as connection:
        # 用run_sync将同步的context.configure适配到异步环境
        await connection.run_sync(lambda conn: context.configure(
            connection=conn, target_metadata=target_metadata
        ))

        await context.run_migrations()

2. 修复空迁移脚本问题 #

空脚本通常是Alembic未检测到模型元数据导致，按以下步骤排查修复：

步骤1：验证模型导入正确性

确认env.py中已正确导入所有模型，且模型继承自同一个Base类：

# env.py中需保留这些导入
from core.database import Base
from models.users.user_model import User
from models.notes.note_model import Note

步骤2：检查数据库连接URL

确保alembic.ini中的sqlalchemy.url为异步格式，与database.py中的连接串一致：

# alembic.ini配置示例
sqlalchemy.url = mysql+asyncmy://你的用户名:你的密码@localhost/notes-app-db

步骤3：强制加载模型元数据

若导入后仍无法检测，可在env.py中手动触发模型元数据加载：

target_metadata = Base.metadata
# 强制加载模型表结构，确保Alembic能识别
User.__table__
Note.__table__

步骤4：重置迁移历史（可选）

如果旧的迁移历史存在损坏，可删除alembic/versions下的所有文件，重新生成初始迁移：

alembic revision --autogenerate -m "Initial migration"
alembic upgrade head

完整修改后的env.py示例 #

from logging.config import fileConfig
from sqlalchemy.ext.asyncio import AsyncEngine
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context
from core.database import Base
from models.notes.note_model import Note
from models.users.user_model import User
import asyncio

config = context.config

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata
# 强制加载模型元数据
User.__table__
Note.__table__


def run_migrations_offline() -> None:
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


async def run_migrations_online() -> None:
    connectable = AsyncEngine(
        engine_from_config(
            config.get_section(config.config_ini_section, {}),
            prefix="sqlalchemy.",
            poolclass=pool.NullPool,
            future=True,
        )
    )

    async with connectable.connect() as connection:
        await connection.run_sync(lambda conn: context.configure(
            connection=conn, target_metadata=target_metadata
        ))

        await context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    asyncio.run(run_migrations_online())

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