FastAPI启动时初始化数据库数据遇AttributeError问题求助

我是FastAPI和SQLAlchemy的新手，尝试在应用启动时初始化部分数据，写了如下代码但启动时无法获取数据库会话，报错AttributeError: __enter__。

实现代码 #

@app.on_event("startup")
async def startup_event():
    with SessionLocal() as session:
        country_dataframe = pd.read_csv('./initialize_data/country.csv')
        for index, row in country_dataframe.iterrows():
            session.add(models.Country(row.to_dict()))
        session.commit()

错误信息 #

ERROR: Traceback (most recent call last): File
"C:\Users\newia\Miniconda3\envs\fastapi\lib\site-packages\starlette\routing.py",
line 540, in lifespan
async for item in self.lifespan_context(app): File "C:\Users\newia\Miniconda3\envs\fastapi\lib\site-packages\starlette\routing.py",
line 481, in default_lifespan
await self.startup() File "C:\Users\newia\Miniconda3\envs\fastapi\lib\site-packages\starlette\routing.py",
line 516, in startup
await handler() File "D:\Software Projects\PythonProjects\LanguageExchange\app.py", line 27, in
startup_event
with SessionLocal() as session: AttributeError: enter

ERROR: Application startup failed. Exiting.

解决方案 #

报错核心原因是异步函数与会话类型不匹配，分两种场景处理：

场景1：使用SQLAlchemy同步会话

如果你的SessionLocal是基于Session类创建的同步会话，启动事件函数不能用async修饰，改成普通函数即可：

@app.on_event("startup")
def startup_event():
    with SessionLocal() as session:
        country_dataframe = pd.read_csv('./initialize_data/country.csv')
        for index, row in country_dataframe.iterrows():
            session.add(models.Country(row.to_dict()))
        session.commit()

场景2：使用SQLAlchemy异步会话

如果你的SessionLocal是基于AsyncSession类创建的异步会话，需要把普通with改成async with，同时保持函数的async修饰，且数据库操作也要用await：

@app.on_event("startup")
async def startup_event():
    async with SessionLocal() as session:
        country_dataframe = pd.read_csv('./initialize_data/country.csv')
        for index, row in country_dataframe.iterrows():
            await session.add(models.Country(row.to_dict()))
        await session.commit()

推荐方案：使用Lifespan上下文管理器

FastAPI推荐用lifespan上下文管理器替代@app.on_event，更符合现代异步应用的生命周期管理规范：

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 应用启动时执行初始化逻辑
    async with SessionLocal() as session:
        country_dataframe = pd.read_csv('./initialize_data/country.csv')
        for index, row in country_dataframe.iterrows():
            await session.add(models.Country(row.to_dict()))
        await session.commit()
    yield
    # 应用关闭时可添加清理逻辑（如关闭资源）
    pass

# 创建应用时指定lifespan
app = FastAPI(lifespan=lifespan)

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