如何在FastAPI+SQLAlchemy+Pydantic(v2)中验证关联ID存在性？

问题描述 #

在FastAPI + SQLAlchemy + Pydantic（v2）项目中，需要验证前端传入的关联模型ID是否存在于数据库，且希望将这类验证融入Pydantic的标准验证流程（复用其错误返回格式），但无法将get_session依赖中的数据库会话（session）传入Pydantic的验证方法，导致验证失败。

示例代码 #

import typing
from typing import Any

import uvicorn
from fastapi import FastAPI, Depends, Body
from pydantic import BaseModel, field_validator, Field, ValidationInfo
from sqlalchemy import ForeignKey, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import declarative_base, Mapped, mapped_column, relationship

Base = declarative_base()
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
SessionLocal = async_sessionmaker(autocommit=False, autoflush=False, bind=engine)

app = FastAPI()

class Question(Base):
    __tablename__ = "questions"
    id: Mapped[int] = mapped_column(primary_key=True)
    text: Mapped[str]
    topic_id: Mapped[int] = mapped_column(ForeignKey("topics.id"))
    topic: Mapped["Topic"] = relationship(back_populates="questions")

class Topic(Base):
    __tablename__ = "topics"
    id: Mapped[int] = mapped_column(primary_key=True)
    text: Mapped[str]
    questions: Mapped[list["Question"]] = relationship(back_populates="topic")

async def get_session():
    db = SessionLocal()
    try:
        yield db
    finally:
        await db.close()

class CreateQuestionDataSchema(BaseModel):
    text: str
    topic: Topic = Field(alias="topic_id", gt=0)

    @field_validator("topic", mode="before")
    @classmethod
    def validate_topic_id(cls, topic_id: int, info: ValidationInfo):
        session = info.context.get("session")  # 无法获取session
        stmt = select(Topic).where(Topic.id == topic_id)
        topic = await session.scalar(stmt)
        if not topic:
            raise ValueError("Topic does not exist")
        return topic

class CreateTopicDataSchema(BaseModel):
    text: str

async def create_db_and_tables():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

@app.on_event("startup")
async def on_startup():
    await create_db_and_tables()

@app.post("/topics")
async def create_topic(text: str = Body(embed=True), session: AsyncSession = Depends(get_session)):
    topic = Topic(text=text)
    session.add(topic)
    await session.commit()
    return topic

@app.post("/things")
async def create_question(data: CreateQuestionDataSchema, session: AsyncSession = Depends(get_session)):
    question = Question(text=data.text, topic=data.topic)
    session.add(question)
    await session.commit()
    return question

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8080, reload=True)

核心痛点 #

class CreateQuestionDataSchema(BaseModel):
    text: str
    topic: Topic = Field(alias="topic_id", gt=0)

    @field_validator("topic", mode="before")
    @classmethod
    def validate_topic_id(cls, topic_id: int, info: ValidationInfo):
        session = info.context.get("session")  # 如何将session传入此处？
        stmt = select(Topic).where(Topic.id == topic_id)
        topic = await session.scalar(stmt)
        if not topic:
            raise ValueError("Topic does not exist")
        return topic

要求：

• 不能通过插入数据验证关联存在性
• 需将关联验证融入Pydantic标准验证流程，复用错误返回格式

解决方案 #

1. 启用Pydantic异步验证 #

Pydantic v2支持异步字段验证器，只需将验证器定义为异步类方法，并设置mode参数为before（适合前置验证ID存在性）。

2. 手动传递Session到Pydantic Context #

在接口函数中，放弃FastAPI自动解析模型的方式，手动调用model_validate方法并将数据库session传入context参数。

3. 优化Schema字段定义 #

直接使用topic_id字段接收前端参数，避免别名混淆，同时保持与数据库字段的一致性。

修改后的完整代码 #

import typing
from typing import Any

import uvicorn
from fastapi import FastAPI, Depends, Body, HTTPException
from pydantic import BaseModel, field_validator, Field, ValidationInfo
from sqlalchemy import ForeignKey, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import declarative_base, Mapped, mapped_column, relationship

Base = declarative_base()
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
SessionLocal = async_sessionmaker(autocommit=False, autoflush=False, bind=engine)

app = FastAPI()

class Question(Base):
    __tablename__ = "questions"
    id: Mapped[int] = mapped_column(primary_key=True)
    text: Mapped[str]
    topic_id: Mapped[int] = mapped_column(ForeignKey("topics.id"))
    topic: Mapped["Topic"] = relationship(back_populates="questions")

class Topic(Base):
    __tablename__ = "topics"
    id: Mapped[int] = mapped_column(primary_key=True)
    text: Mapped[str]
    questions: Mapped[list["Question"]] = relationship(back_populates="topic")

async def get_session():
    db = SessionLocal()
    try:
        yield db
    finally:
        await db.close()

class CreateQuestionDataSchema(BaseModel):
    text: str
    topic_id: int = Field(gt=0)

    @field_validator("topic_id", mode="before")
    @classmethod
    async def validate_topic_exists(cls, topic_id: int, info: ValidationInfo):
        session: AsyncSession = info.context.get("session")
        if not session:
            raise ValueError("Database session not available")
        
        stmt = select(Topic).where(Topic.id == topic_id)
        topic = await session.scalar(stmt)
        if not topic:
            raise ValueError(f"Topic with id {topic_id} does not exist")
        return topic_id

class CreateTopicDataSchema(BaseModel):
    text: str

async def create_db_and_tables():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

@app.on_event("startup")
async def on_startup():
    await create_db_and_tables()

@app.post("/topics")
async def create_topic(text: str = Body(embed=True), session: AsyncSession = Depends(get_session)):
    topic = Topic(text=text)
    session.add(topic)
    await session.commit()
    await session.refresh(topic)
    return topic

@app.post("/things")
async def create_question(session: AsyncSession = Depends(get_session), payload: dict = Body(...)):
    try:
        # 手动验证并解析数据，传入session到context
        data = CreateQuestionDataSchema.model_validate(payload, context={"session": session})
    except ValueError as e:
        # 转换为FastAPI标准422错误，保持统一格式
        raise HTTPException(status_code=422, detail=str(e))
    
    question = Question(text=data.text, topic_id=data.topic_id)
    session.add(question)
    await session.commit()
    await session.refresh(question)
    return question

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8080, reload=True)

关键说明 #

• 异步验证器：用async修饰验证方法，支持异步SQLAlchemy查询
• Context传递Session：通过model_validate的context参数将session传入验证器，解决依赖注入无法直接传递到Pydantic的问题
• 错误格式复用：捕获Pydantic验证错误并转换为FastAPI标准的422响应，与默认验证错误格式一致
• 批量验证扩展：可封装通用验证函数，减少多关联场景下的重复代码：

  async def validate_model_exists(session: AsyncSession, model, model_id: int) -> bool:
      stmt = select(model).where(model.id == model_id)
      return await session.scalar(stmt) is not None
  

内容的提问来源于stack exchange，提问作者Альберт Александров