FastAPI接口参数校验与异常处理

FastAPI 的一个重要优势是基于类型标注和 Pydantic 做参数校验。接口参数写清楚后，框架可以自动解析请求、校验字段、生成 OpenAPI 文档，并在参数错误时返回结构化响应。

本文通过一个用户创建接口，介绍路径参数、查询参数、请求体校验、自定义异常和统一错误返回的实践方式。

基础项目结构

一个简单项目可以这样组织：

app/
  main.py
  schemas.py
  exceptions.py

安装依赖：

1	pip install fastapi uvicorn

启动命令：

1	uvicorn app.main:app --reload

访问 http://127.0.0.1:8000/docs 可以查看自动生成的接口文档。

定义请求体模型

在 schemas.py 中定义用户创建请求：

from pydantic import BaseModel, EmailStr, Field

class CreateUserRequest(BaseModel):
    username: str = Field(min_length=3, max_length=32)
    email: EmailStr
    age: int = Field(ge=0, le=150)

字段含义：

username 长度必须在 3 到 32 之间。
email 必须是合法邮箱格式。
age 必须在 0 到 150 之间。

如果使用 EmailStr，需要安装邮箱校验依赖：

1	pip install "pydantic[email]"

响应模型也建议显式定义：

class UserResponse(BaseModel):
    id: int
    username: str
    email: EmailStr

路径参数和查询参数

在 main.py 中创建接口：

from fastapi import FastAPI, Query

from .schemas import CreateUserRequest, UserResponse

app = FastAPI()

@app.get("/users/{user_id}", response_model=UserResponse)
def get_user(
    user_id: int,
    include_deleted: bool = Query(default=False),
):
    return {
        "id": user_id,
        "username": "demo",
        "email": "demo@example.com",
    }

user_id 会自动从路径中解析为整数。如果请求 /users/abc，FastAPI 会返回参数错误。

查询参数 include_deleted 会自动解析布尔值，例如：

1	curl 'http://127.0.0.1:8000/users/1?include_deleted=true'

请求体校验

创建用户接口：

@app.post("/users", response_model=UserResponse)
def create_user(req: CreateUserRequest):
    return {
        "id": 1,
        "username": req.username,
        "email": req.email,
    }

测试请求：

1
2
3

curl -X POST 'http://127.0.0.1:8000/users' \
  -H 'Content-Type: application/json' \
  -d '{"username":"clang","email":"clang@example.com","age":20}'

如果 username 过短或邮箱格式错误，FastAPI 会返回 422，并说明具体字段错误。

自定义业务异常

参数校验解决的是输入格式问题，业务错误还需要自己定义。例如用户不存在、用户名重复、权限不足。

在 exceptions.py 中定义：

class AppError(Exception):
    def __init__(self, code: str, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code

在入口注册异常处理器：

from fastapi import Request
from fastapi.responses import JSONResponse

from .exceptions import AppError

@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "code": exc.code,
            "message": exc.message,
        },
    )

业务代码中抛出：

@app.get("/users/{user_id}", response_model=UserResponse)
def get_user(user_id: int):
    if user_id == 404:
        raise AppError("USER_NOT_FOUND", "user not found", 404)

    return {
        "id": user_id,
        "username": "demo",
        "email": "demo@example.com",
    }

这样业务错误的返回结构就统一了。

覆盖参数校验错误格式

如果希望 422 参数错误也返回统一结构，可以处理 RequestValidationError：

from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_error_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "code": "VALIDATION_ERROR",
            "message": "request validation failed",
            "details": exc.errors(),
        },
    )

details 中包含字段位置、错误类型和错误原因。对前端联调来说，这比只返回一段字符串更好处理。

不要吞掉未知异常

可以增加全局异常处理器，但不要把未知异常伪装成业务错误：

import logging

logger = logging.getLogger(__name__)

@app.exception_handler(Exception)
async def unknown_error_handler(request: Request, exc: Exception):
    logger.exception("unhandled error path=%s", request.url.path)
    return JSONResponse(
        status_code=500,
        content={
            "code": "INTERNAL_ERROR",
            "message": "internal server error",
        },
    )