执行Alembic迁移遇MissingGreenlet错误，求解决办法

1. 修复模型关系定义的逻辑矛盾 #

你的Group和User关系存在匹配错误：

• Group表通过user_id外键关联User，说明一个Group属于一个User，但当前Group的user关系定义为List["User"]（一对多），和User的group（一对一）完全冲突。

修正后的模型代码：

class Group(Base):
    __tablename__ = 'group'

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(nullable=False, index=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    user_id: Mapped[int] = mapped_column(ForeignKey('user.id'))
    # 一个Group对应一个User，关系类型改为单个User
    user: Mapped["User"] = relationship(back_populates="groups")


class User(Base):
    __tablename__ = "user"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(nullable=False)
    surname: Mapped[str]
    username: Mapped[str] = mapped_column(index=True)
    phone: Mapped[str]
    email: Mapped[str] = mapped_column(unique=True)
    role: Mapped[Role] = mapped_column(nullable=True)
    # 一个User对应多个Group，关系类型改为List[Group]
    groups: Mapped[List["Group"]] = relationship(back_populates="user", lazy="selectin")
    image: Mapped[str] = mapped_column(nullable=True, default='image.jpg')
    is_blocked: Mapped[bool]
    created_at: Mapped[datetime] = mapped_column(server_default=func.now(), nullable=True)
    modified_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), onupdate=func.now(), nullable=True
    )

2. 替换同步时间默认值 #

把default=datetime.now替换为server_default=func.now()，避免在异步上下文执行同步IO操作，减少潜在的greenlet冲突。

3. 配置Alembic适配异步引擎 #

Alembic默认是同步模式，需要修改alembic/env.py来支持异步SQLAlchemy：

步骤1：导入异步相关模块 #

from sqlalchemy.ext.asyncio import AsyncEngine
from your_project.db_settings import sessionmanager  # 替换为你的项目路径

步骤2：修改离线迁移函数（如果使用） #

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

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

步骤3：改写在线迁移为异步版本 #

async def run_migrations_online() -> None:
    connectable = sessionmanager._engine

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()


def do_run_migrations(connection):
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        expire_on_commit=False
    )

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

步骤4：调整脚本入口调用异步函数 #

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

4. 验证迁移命令 #

配置完成后，正常执行迁移命令即可：

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

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