3. FastAPI核心基础:路径操作、请求方法、路径参数、查询参数、请求体、响应模型、状态码与异常处理

FastAPI 之所以能在短时间内成为 Python 后端开发的主流框架,核心在于它将现代类型系统与 Web 协议规范进行了深度绑定。本章将围绕构建一个 API 端点所需的最小必要知识展开:从路径操作与请求方法的声明,到参数提取的三种主要方式(路径、查询、请求体),再到响应模型与状态码的规范化输出,最后通过异常处理机制构建健壮的容错边界。理解这些基础概念之间的协作关系,是后续掌握依赖注入、安全认证、数据库集成等高级特性的前提。

FastAPI 请求-响应核心链路 HTTP 客户端 (浏览器/移动端) 请求方法 GET / POST PUT / DELETE PATCH / OPTIONS 路径操作装饰器 @app.get("/items") @app.post("/items") 参数提取层 路径参数: /items/{item_id} 查询参数: ?skip=0&limit=10 请求体: JSON Body 路径操作函数 async def handler() 业务逻辑处理 响应模型 response_model= Pydantic Schema 状态码与异常 200/201/404 HTTPException FastAPI 通过类型声明自动完成参数校验、序列化与文档生成

3.1 路径操作与请求方法:定义API的入口

3.1.1 核心原理:装饰器驱动的路由注册

FastAPI 的路由系统基于 Python 装饰器实现。每个路径操作(Path Operation)由三要素组成:路径(Path)、请求方法(Method)和处理函数(Function)。当客户端发起请求时,FastAPI 内部通过路由表匹配请求的 URL 与 HTTP 方法,找到对应的处理函数并执行。这种设计将路由声明与业务逻辑解耦,使得代码结构清晰且易于维护。

路径操作装饰器不仅声明了路由,还隐式地定义了 API 的接口契约。例如,@app.post("/items") 表示该端点接受 POST 方法,通常用于创建资源。FastAPI 会自动为每个端点生成 OpenAPI 文档,这得益于装饰器对函数签名和类型注解的深度解析。

3.1.2 操作流程:声明一个基础端点

from fastapi import FastAPI

app = FastAPI()

@app.get("/health")
async def health_check():
    return {"status": "ok"}

@app.post("/items")
async def create_item():
    return {"message": "Item created"}

上述代码展示了最基本的路径操作声明。需要注意:函数名可以任意,但路径和请求方法决定了端点的唯一性。FastAPI 支持所有标准 HTTP 方法:GET、POST、PUT、DELETE、PATCH、OPTIONS、HEAD。选择合适的方法遵循 RESTful 语义:GET 用于查询,POST 用于创建,PUT 用于整体更新,PATCH 用于部分更新,DELETE 用于删除。

3.1.3 案例拆解:订单服务的方法设计

假设构建一个订单服务,合理的端点设计如下:

请求方法路径功能状态码
GET/orders获取订单列表200
POST/orders创建新订单201
GET/orders/{order_id}获取订单详情200
PUT/orders/{order_id}整体更新订单200
DELETE/orders/{order_id}删除订单204

这种设计遵循了资源导向的架构风格。在实际项目中,如果发现某个端点同时需要 GET 和 POST 才能完成业务,通常意味着资源划分不够细致,需要重新审视 API 设计。

常见误区:不要将动作写在路径中,例如 /orders/create 或 /orders/delete。正确的做法是使用 HTTP 方法表达动作语义,路径只表示资源。这有助于保持 API 的整洁性和可维护性。

3.2 路径参数与查询参数:从URL中提取数据

3.2.1 核心原理:类型声明驱动的自动转换

路径参数是 URL 路径中可变的部分,用花括号 {} 标识。查询参数是 URL 中 ? 后面的键值对。FastAPI 通过函数参数的类型注解自动完成参数提取、类型转换和校验。例如,声明 item_id: int 时,FastAPI 会自动将路径中的字符串转换为整数,并在转换失败时返回 422 校验错误。

3.2.2 操作流程:参数声明与默认值

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(
    item_id: int,
    q: str | None = None,
    skip: int = Query(0, ge=0),
    limit: int = Query(10, le=100)
):
    return {"item_id": item_id, "q": q, "skip": skip, "limit": limit}

在上述代码中:item_id 是路径参数,必须提供;q 是可选查询参数,默认值为 None;skip 和 limit 使用 Query 类添加了额外的校验约束(skip ≥ 0,limit ≤ 100)。FastAPI 会自动生成对应的 OpenAPI 文档,并在请求参数不满足约束时返回 422 错误。

3.2.3 参数优先级与类型转换规则

当函数参数同时包含路径参数和查询参数时,FastAPI 根据参数名与路径模板的匹配情况自动区分。如果参数名与路径中的 {param} 一致,则视为路径参数;否则视为查询参数。这种机制避免了手动指定参数来源的繁琐操作。

实战建议:对于分页参数,建议始终设置默认值(如 skip=0, limit=10),并添加合理的上限约束(如 limit ≤ 100)。这可以防止客户端请求过大数据量导致服务端资源耗尽。

3.3 请求体与响应模型:结构化数据交互

3.3.1 核心原理:Pydantic 模型的桥接作用

请求体(Request Body)是客户端发送给服务端的数据,通常为 JSON 格式。FastAPI 使用 Pydantic 模型定义请求体的结构,自动完成 JSON 解析、字段校验和类型转换。响应模型(Response Model)则定义了服务端返回数据的结构,FastAPI 会根据响应模型过滤掉多余字段,确保输出符合契约。

3.3.2 操作流程:定义模型与端点

from pydantic import BaseModel, Field
from fastapi import FastAPI

app = FastAPI()

class ItemCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=50)
    price: float = Field(..., gt=0)
    tax: float | None = None

class ItemResponse(BaseModel):
    name: str
    price: float
    tax: float | None = None

@app.post("/items", response_model=ItemResponse, status_code=201)
async def create_item(item: ItemCreate):
    # 业务逻辑:保存到数据库等
    return {"name": item.name, "price": item.price, "tax": item.tax}

在这个例子中,ItemCreate 作为请求体模型,ItemResponse 作为响应模型。FastAPI 会:1) 校验请求体是否符合 ItemCreate 的约束;2) 将处理函数的返回值按照 ItemResponse 进行序列化输出。如果返回值包含 ItemResponse 中未定义的字段,这些字段会被自动过滤。

3.3.3 响应模型的高级用法:嵌套与别名

特性说明示例
嵌套模型模型可以包含子模型class Order(BaseModel): items: List[Item]
字段别名使用 alias 映射不同名称class User(BaseModel): user_name: str = Field(alias="username")
排除字段使用 response_model_excluderesponse_model_exclude={"secret"}
包含字段使用 response_model_includeresponse_model_include={"name", "price"}

响应模型的核心价值在于契约安全。当 API 演进时,如果内部数据结构发生变化,只要响应模型不变,客户端就不会受到影响。这为微服务架构中的接口兼容性提供了保障。

经验提示:建议为每个端点单独定义请求模型和响应模型,而不是直接复用数据库模型。数据库模型通常包含内部字段(如 created_at, updated_at),这些字段不应暴露给客户端。通过响应模型进行字段过滤,可以有效防止数据泄露。

3.4 状态码与异常处理:构建健壮的API

3.4.1 核心原理:HTTP状态码的语义化使用

HTTP 状态码是服务端向客户端传达请求处理结果的标准方式。FastAPI 允许在装饰器中使用 status_code 参数指定默认状态码,也可以在处理函数中动态返回 Response 对象或抛出 HTTPException 来覆盖默认状态码。正确使用状态码可以显著提升 API 的可理解性和调试效率。

3.4.2 操作流程:异常处理与自定义响应

from fastapi import FastAPI, HTTPException, status
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id <= 0:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Item ID must be positive",
            headers={"X-Error": "Invalid ID"}
        )
    if item_id == 999:
        return JSONResponse(
            status_code=status.HTTP_404_NOT_FOUND,
            content={"message": "Item not found"}
        )
    return {"item_id": item_id, "name": f"Item {item_id}"}

在上述代码中,HTTPException 用于抛出带状态码和错误详情的异常,JSONResponse 用于完全自定义响应内容。FastAPI 还支持注册全局异常处理器,用于捕获未处理的异常并统一格式化输出。

3.4.3 状态码选择速查表

状态码含义适用场景
200OKGET 请求成功
201CreatedPOST 创建资源成功
204No ContentDELETE 删除成功,无返回体
400Bad Request参数校验失败
401Unauthorized未认证或认证失败
403Forbidden已认证但无权限
404Not Found资源不存在
422Unprocessable Entity请求体校验失败
500Internal Server Error服务端未处理异常

3.4.4 案例拆解:全局异常处理器的实现

在一个电商系统中,数据库连接异常是常见问题。如果不做全局处理,客户端会收到 500 错误且响应体为 HTML 格式,这对前端解析极不友好。通过注册全局异常处理器,可以统一返回 JSON 格式的错误信息:

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(
        status_code=400,
        content={"code": "INVALID_VALUE", "message": str(exc)}
    )

@app.exception_handler(Exception)
async def generic_error_handler(request: Request, exc: Exception):
    # 记录日志到监控系统
    logger.error(f"Unhandled error: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"code": "INTERNAL_ERROR", "message": "Internal server error"}
    )

这种设计将异常处理逻辑集中管理,避免了在每个端点中重复编写 try-except 代码。同时,通过日志记录,可以快速定位生产环境中的问题。

风险提示:在全局异常处理器中,不要将原始异常信息直接返回给客户端。这可能会泄露内部实现细节(如数据库连接字符串、文件路径等),给攻击者提供可利用信息。应记录详细日志,但对外只返回通用错误消息。

3.5 综合实战:构建一个完整的商品管理API

3.5.1 场景与约束

假设需要为一个小型电商平台构建商品管理 API。约束条件:商品名称唯一,价格必须大于 0,库存为非负整数。客户端需要能够创建商品、查询商品列表(支持分页)、获取单个商品详情、更新商品信息、删除商品。

3.5.2 实现方案

from fastapi import FastAPI, HTTPException, Query, status
from pydantic import BaseModel, Field
from typing import List, Optional

app = FastAPI()

# 内存数据库
items_db = {}
next_id = 1

class ItemCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=50)
    price: float = Field(..., gt=0)
    stock: int = Field(..., ge=0)

class ItemUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=50)
    price: Optional[float] = Field(None, gt=0)
    stock: Optional[int] = Field(None, ge=0)

class ItemResponse(BaseModel):
    id: int
    name: str
    price: float
    stock: int

@app.post("/items", response_model=ItemResponse, status_code=status.HTTP_201_CREATED)
async def create_item(item: ItemCreate):
    global next_id
    # 检查名称唯一性
    for existing in items_db.values():
        if existing["name"] == item.name:
            raise HTTPException(status_code=400, detail="Item name already exists")
    item_id = next_id
    items_db[item_id] = {"id": item_id, **item.model_dump()}
    next_id += 1
    return items_db[item_id]

@app.get("/items", response_model=List[ItemResponse])
async def list_items(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100)
):
    items = list(items_db.values())
    return items[skip:skip + limit]

@app.get("/items/{item_id}", response_model=ItemResponse)
async def get_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return items_db[item_id]

@app.put("/items/{item_id}", response_model=ItemResponse)
async def update_item(item_id: int, item: ItemUpdate):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    current = items_db[item_id]
    update_data = item.model_dump(exclude_unset=True)
    if "name" in update_data:
        for existing in items_db.values():
            if existing["name"] == update_data["name"] and existing["id"] != item_id:
                raise HTTPException(status_code=400, detail="Item name already exists")
    current.update(update_data)
    return current

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    del items_db[item_id]
    return None

3.5.3 案例复盘与关键决策

在这个案例中,有几个关键设计决策值得总结:

  • 使用 model_dump(exclude_unset=True) 实现部分更新,只更新客户端显式提供的字段,避免覆盖未修改的数据。
  • 状态码的精确使用:创建返回 201,删除返回 204,校验失败返回 400,资源不存在返回 404。这符合 RESTful API 的最佳实践。
  • 响应模型隔离:内部存储使用字典,但对外输出通过 ItemResponse 模型,确保字段一致性。
  • 内存存储的局限性:此案例使用内存字典模拟数据库,仅用于教学演示。生产环境应使用真实数据库,并考虑并发访问的数据一致性。
性能考量:对于列表接口,分页参数 skip 和 limit 的校验非常重要。如果不加限制,客户端可以请求 limit=1000000,导致服务端内存溢出。通过 le=100 约束,将单次请求的数据量限制在合理范围内。

3.6 常见误区与总结复盘

3.6.1 误区一:混淆路径参数与查询参数

路径参数用于标识特定资源(如 /users/42),查询参数用于过滤、排序或分页(如 /users?role=admin)。将资源标识放在查询参数中(如 /users?id=42)虽然技术上可行,但不符合 RESTful 规范,且会降低 URL 的可读性。

3.6.2 误区二:忽略响应模型的过滤作用

如果处理函数返回的字典包含敏感字段(如密码哈希、内部 ID),而响应模型未排除这些字段,数据将直接暴露给客户端。始终使用响应模型作为 API 的对外契约,并定期审查模型字段。

3.6.3 误区三:异常处理过于笼统

将所有异常都捕获并返回 500 错误,会掩盖真实的业务逻辑错误。应该区分可预期的业务异常(使用 HTTPException 抛出)和不可预期的系统异常(记录日志并返回 500)。

3.6.4 总结:核心知识框架

组件核心作用关键参数/方法
路径操作装饰器注册路由@app.get/post/put/delete
路径参数标识资源{param} 语法
查询参数过滤/分页Query 类约束
请求体模型解析与校验Pydantic BaseModel
响应模型序列化与过滤response_model 参数
状态码表达处理结果status_code 参数
异常处理容错与错误反馈HTTPException, exception_handler

掌握这些核心基础后,你已经能够构建结构清晰、类型安全、文档自动化的 RESTful API。下一阶段的学习将围绕更高级的主题展开,包括依赖注入、数据库集成、安全认证、文件上传、WebSocket 等。建议在继续学习之前,通过实际项目反复练习本章内容,确保对请求-响应链路的每个环节都有深刻理解。