.webp){kind=small}
FastAPI 系列
数据库会话怎么进接口（当前篇）
数据库代码一旦写进路由函数，测试就只能连真实库。正确形状是：引擎和会话工厂在应用启动时创建，每个请求通过 Depends 拿到独立会话，测试时整段替换。
会话依赖
from typing import Annotated, Generator
from fastapi import Depends
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
engine = create_engine("sqlite:///./app.db", connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
def get_db() -> Generator[Session, None, None]:
db = SessionLocal()
try:
yield db
db.commit()
except Exception:
db.rollback()
raise
finally:
db.close()
DbDep = Annotated[Session, Depends(get_db)]
路径函数只声明 db: DbDep。成功则提交，异常则回滚。不要在三个路由里复制这 15 行。
路由里怎么用
from fastapi import APIRouter, HTTPException
from app.models import Item
router = APIRouter(prefix="/items", tags=["items"])
@router.get("/{item_id}")
def read_item(item_id: int, db: DbDep):
item = db.get(Item, item_id)
if item is None:
raise HTTPException(status_code=404, detail="Item not found")
return item
这里用同步 def 是因为同步 SQLAlchemy 驱动会阻塞。要么换 AsyncSession 和异步驱动，要么接受 FastAPI 把 def 放到线程池。不要在 async def 里调用同步 db.query。
异步会话的形状
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
engine = create_async_engine(url)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
async def get_db():
async with SessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
异步引擎的 URL 形如 postgresql+asyncpg://...。混用同步 URL 和 AsyncSession 会在运行时才爆。
测试时怎么不连真实库
from fastapi.testclient import TestClient
from app.main import app
from app.db import get_db
def fake_db():
yield MemorySession()
app.dependency_overrides[get_db] = fake_db
client = TestClient(app)
覆盖的是函数对象 get_db 本身，不是字符串名字。测试结束后清空 app.dependency_overrides。
常见误区
模块级全局 session
导入时创建一次，所有请求共用，必出并发污染。
在依赖里 commit，在路由里再 commit
选一处作为事务边界。推荐依赖在成功路径 commit，路由只 flush 或添加对象。
把连接字符串写死在代码里
用环境变量。部署篇会把配置和密钥从代码里拿出去。
小结
会话是请求级资源：yield 借出，结束归还。路由只管「查哪一行、改哪一行」。下一篇把失败从 raise Exception 收成稳定的 HTTP 错误。
参考资料
数据库会话怎么进接口：SQLAlchemy 与依赖覆盖
https://lautung.com/archives/fastapi-07-database
评论