FastAPI中使用异步SQLAlchemy执行alembic upgrade head报错求解

问题背景 #

命令行执行alembic upgrade head可正常升级数据库架构，但希望在FastAPI启动时自动完成该操作，编写了以下异步方法：

async def run_migrations():
    alembic_cfg = Config("alembic.ini")
    url = str(engine.url)
    async with engine.begin() as conn:
        await conn.run_sync(alembic_cfg.set_main_option, "sqlalchemy.url", url)
        await conn.run_sync(command.upgrade, alembic_cfg, "head")

在FastAPI的lifespan启动阶段调用该方法时，出现错误：

TypeError: Config.set_main_option() takes 3 positional arguments but 4 were given

尝试修改写法：

await conn.run_sync(alembic_cfg.set_main_option("sqlalchemy.url", url))

又出现错误：

TypeError: 'NoneType' object is not callable

使用依赖版本：

• SQLAlchemy 2.0.32
• uvicorn 0.30.6
• fastapi 0.112.1
• alembic 1.13.2

问题原因与解决方案 #

错误原因 #

conn.run_sync()会自动将当前的同步连接对象作为第一个参数传递给传入的函数，但Config.set_main_option()是实例方法，本身已经绑定了self（即alembic_cfg），此时run_sync再传入连接对象就导致参数数量超出预期。

正确写法 #

将设置配置和迁移的逻辑封装进同步函数，再通过run_sync调用：

async def run_migrations():
    alembic_cfg = Config("alembic.ini")
    url = str(engine.url)
    
    async with engine.begin() as conn:
        def run_sync_migrations(sync_conn):
            # 修改配置中的数据库链接
            alembic_cfg.set_main_option("sqlalchemy.url", url)
            # 绑定当前同步连接到配置，避免重复创建连接
            alembic_cfg.attributes['connection'] = sync_conn
            # 执行升级命令
            command.upgrade(alembic_cfg, "head")
        
        await conn.run_sync(run_sync_migrations)

更简便的替代方案 #

如果启动阶段不需要严格异步执行，可直接在FastAPI实例创建前同步执行迁移：

from alembic.config import Config
from alembic import command

def run_migrations_sync():
    alembic_cfg = Config("alembic.ini")
    command.upgrade(alembic_cfg, "head")

# 在FastAPI启动前执行迁移
run_migrations_sync()

app = FastAPI()

# 若需在startup事件中执行，仍采用封装同步函数的异步写法
@app.on_event("startup")
async def startup_event():
    await run_migrations()

关键说明 #

• run_sync()要求传入可调用对象，且会自动把同步连接作为第一个参数传入，因此直接传实例方法会引发参数冲突，必须用普通函数封装逻辑。
• 将同步连接绑定到alembic_cfg.attributes['connection']，可让Alembic复用当前数据库连接，避免额外建立连接开销。

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