FastAPI异步数据库依赖实现报错：/users路由返回422错误

问题现象 #

访问FastAPI项目的/users路由时，返回422 Unprocessable Entity错误，移除数据库依赖后路由可正常运行。

错误日志：

127.0.0.1:35352 - "GET /users/ HTTP/1.1" 422 Unprocessable Entity
2025-01-14 20:51

返回的错误详情：

{
  "detail": [
    {
      "type": "missing",
      "loc": [
        "query",
        "args"
      ],
      "msg": "Field required",
      "input": null
    },
    {
      "type": "missing",
      "loc": [
        "query",
        "kwargs"
      ],
      "msg": "Field required",
      "input": null
    }
  ]
}

相关代码 #

路由方法代码 #

router = APIRouter(
    prefix="/users",
    tags=["users"]
)

@router.get("/", response_model=List[schemas.UserDetail])
async def list_user(db: AsyncSession = Depends(async_db_session_dependency)):
    try:
        users = await crud.get_users(db)
        if not users:
            raise HTTPException(status_code=404, detail="No users found")
        validated_users = [schemas.UserDetail.model_validate(user) for user in users]
        
        return validated_users
    
    except Exception as e:
        print("Error: ", e)
        raise HTTPException(
            status_code=400, 
            detail="Error getting users"
        )

数据库依赖代码 #

from typing import Annotated, AsyncGenerator, Generator
from fastapi import Depends
from sqlalchemy.orm import Session
from sqlalchemy.ext.asyncio import AsyncSession
from contextlib import asynccontextmanager

from services.database import SessionLocal, AsyncSessionLocal

# Dependency for synchronous session
def get_db_session() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Dependency for asynchronous session
async def get_async_db_session() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        yield session

db_session_dependency = Annotated[Session, Depends(get_db_session)]
async_db_session_dependency = Annotated[AsyncSession, Depends(get_async_db_session)]

问题分析 #

错误提示要求查询参数中提供args和kwargs，核心原因是路由函数对数据库依赖的写法错误：async_db_session_dependency已经是包含Depends(get_async_db_session)的Annotated类型，却再次用Depends()包裹，导致FastAPI误将其解析为需要接收调用参数的可调用对象，进而要求从查询参数获取args和kwargs。

解决方案 #

修改路由函数的参数声明，直接使用定义好的async_db_session_dependency类型，无需嵌套Depends()：

@router.get("/", response_model=List[schemas.UserDetail])
async def list_user(db: async_db_session_dependency):
    try:
        users = await crud.get_users(db)
        if not users:
            raise HTTPException(status_code=404, detail="No users found")
        validated_users = [schemas.UserDetail.model_validate(user) for user in users]
        
        return validated_users
    
    except Exception as e:
        print("Error: ", e)
        raise HTTPException(
            status_code=400, 
            detail="Error getting users"
        )

调整后FastAPI会正确识别数据库依赖，从异步生成器中获取数据库会话，不再要求查询参数中的args和kwargs。

内容的提问来源于stack exchange，提问作者O.G Bellz