FastAPI中Pydantic自定义校验：如何访问请求数据（如HTTP Header）？

需求说明 #

需要在FastAPI的POST /items/接口中，校验请求体里的item.name必须与请求的Host Header值完全一致，同时解决Pydantic校验器如何访问FastAPI请求数据的问题。

实现方式 #

方式一：路由内直接校验（最简单） #

直接在路由函数中获取请求对象，拿到Host后完成校验，无需修改原有模型结构：

from dataclasses import dataclass
from typing import Union
from fastapi import FastAPI, Request, HTTPException

@dataclass
class Item:
    name: str
    price: float
    description: Union[str, None] = None
    tax: Union[float, None] = None

app = FastAPI()

@app.post("/items/")
async def create_item(request: Request, item: Item):
    host = request.headers.get("Host")
    # 校验Host存在且name与Host一致
    if not host or item.name != host:
        raise HTTPException(
            status_code=400,
            detail=f"name必须与Host Header值一致，当前Host为{host}"
        )
    return item

适用场景：单路由校验，快速实现，逻辑直观。

方式二：自定义依赖复用校验逻辑 #

将校验逻辑抽成独立的依赖函数，方便多个路由复用：

from dataclasses import dataclass
from typing import Union
from fastapi import FastAPI, Request, HTTPException, Depends

@dataclass
class Item:
    name: str
    price: float
    description: Union[str, None] = None
    tax: Union[float, None] = None

# 自定义依赖：校验item.name与Host一致
async def validate_item_host(request: Request, item: Item):
    host = request.headers.get("Host")
    if not host or item.name != host:
        raise HTTPException(
            status_code=400,
            detail=f"name必须与Host Header值一致，当前Host为{host}"
        )
    return item

app = FastAPI()

@app.post("/items/")
async def create_item(validated_item: Item = Depends(validate_item_host)):
    return validated_item

适用场景：多个路由需要相同校验规则时，通过依赖复用逻辑，保持路由代码简洁。

方式三：Pydantic模型内置校验（模型驱动） #

如果希望校验规则与模型绑定，可改用Pydantic BaseModel（替代dataclass），通过校验器结合请求上下文传递Host：

from pydantic import BaseModel, field_validator, ValidationInfo
from typing import Union
from fastapi import FastAPI, Request, HTTPException

class Item(BaseModel):
    name: str
    price: float
    description: Union[str, None] = None
    tax: Union[float, None] = None

    # 字段校验器：校验name与Host一致
    @field_validator('name')
    def name_matches_host(cls, value, info: ValidationInfo):
        host = info.context.get('host')
        if not host or value != host:
            raise ValueError(f"name必须与Host Header值一致，当前Host为{host}")
        return value

app = FastAPI()

@app.post("/items/")
async def create_item(request: Request):
    host = request.headers.get("Host")
    if not host:
        raise HTTPException(status_code=400, detail="缺少Host Header")
    
    # 解析请求体并传入Host上下文完成校验
    item_data = await request.json()
    try:
        item = Item.model_validate(item_data, context={'host': host})
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    
    return item

适用场景：需要模型自身携带校验规则，或校验逻辑与模型强关联的场景。

关键说明 #

• Pydantic原生校验器默认无法直接访问FastAPI请求数据，需通过请求对象传递或依赖注入的方式将请求上下文（如Host）传入校验逻辑。
• 本场景无外部IO调用，符合Pydantic校验器的使用规范。

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