如何在FastAPI中记录Pydantic类型校验失败的服务端日志？

问题场景 #

当FastAPI接收到不符合Pydantic模型定义的请求时，会返回422 Unprocessable Entity错误，但这类错误默认仅反馈给客户端，服务端不会自动记录相关日志。比如以下示例：

示例代码 #

from pydantic import BaseModel
from fastapi import FastAPI, status

app = FastAPI()

class StudentModel(BaseModel):
    id: int
    name: str
    surname: str
    
@app.post("/", status_code=status.HTTP_201_CREATED)
async def create_student(student: StudentModel):
    # 业务逻辑实现
    pass

错误请求 #

{
  "name": "John",
  "surname": "Smith"
}

此时服务端返回422错误，但无日志留存，我们需要在服务端捕获并记录这类错误的请求详情。

解决方案 #

方法1：自定义RequestValidationError异常处理器 #

FastAPI允许针对RequestValidationError（Pydantic验证失败触发的异常）编写自定义处理器，在其中添加日志记录逻辑：

from fastapi import FastAPI, status, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
import logging

# 配置日志输出级别和格式
logging.basicConfig(
    level=logging.ERROR,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

app = FastAPI()

class StudentModel(BaseModel):
    id: int
    name: str
    surname: str

@app.exception_handler(RequestValidationError)
async def validation_error_logger(request: Request, exc: RequestValidationError):
    # 记录错误详情：请求URL、原始请求体、验证错误信息
    logger.error(
        f"请求验证失败 | URL: {request.url} | 请求体: {await request.body().decode('utf-8')} | 错误详情: {exc.errors()}"
    )
    # 保留原有422响应结构，不影响客户端接收错误信息
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={"detail": exc.errors(), "body": exc.body}
    )

@app.post("/", status_code=status.HTTP_201_CREATED)
async def create_student(student: StudentModel):
    pass

方法2：使用全局中间件捕获异常 #

如果需要统一处理更多类型的请求异常，可通过Starlette中间件实现全局日志捕获：

from fastapi import FastAPI, status
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
import logging
from fastapi.exceptions import RequestValidationError

# 配置日志
logging.basicConfig(
    level=logging.ERROR,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

app = FastAPI()

class GlobalErrorLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        try:
            response = await call_next(request)
            return response
        except Exception as exc:
            # 仅针对RequestValidationError记录日志
            if isinstance(exc, RequestValidationError):
                logger.error(
                    f"请求验证失败 | URL: {request.url} | 请求体: {await request.body().decode('utf-8')} | 错误详情: {exc.errors()}"
                )
            # 重新抛出异常，让默认或自定义处理器处理响应
            raise exc

app.add_middleware(GlobalErrorLoggingMiddleware)

class StudentModel(BaseModel):
    id: int
    name: str
    surname: str

@app.post("/", status_code=status.HTTP_201_CREATED)
async def create_student(student: StudentModel):
    pass

注意事项 #

• 两种方法都能完整记录错误请求的关键信息，便于排查问题
• 异常处理器方式更精准针对Pydantic验证错误，逻辑更聚焦
• 中间件方式适合全局统一日志管理，可扩展处理其他类型异常

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