3. FastAPI核心基础:路径操作、请求方法、路径参数、查询参数、请求体、响应模型、状态码与异常处理
FastAPI 之所以能在短时间内成为 Python 后端开发的主流框架,核心在于它将现代类型系统与 Web 协议规范进行了深度绑定。本章将围绕构建一个 API 端点所需的最小必要知识展开:从路径操作与请求方法的声明,到参数提取的三种主要方式(路径、查询、请求体),再到响应模型与状态码的规范化输出,最后通过异常处理机制构建健壮的容错边界。理解这些基础概念之间的协作关系,是后续掌握依赖注入、安全认证、数据库集成等高级特性的前提。
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} 一致,则视为路径参数;否则视为查询参数。这种机制避免了手动指定参数来源的繁琐操作。
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_exclude | response_model_exclude={"secret"} |
| 包含字段 | 使用 response_model_include | response_model_include={"name", "price"} |
响应模型的核心价值在于契约安全。当 API 演进时,如果内部数据结构发生变化,只要响应模型不变,客户端就不会受到影响。这为微服务架构中的接口兼容性提供了保障。
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 状态码选择速查表
| 状态码 | 含义 | 适用场景 |
|---|---|---|
| 200 | OK | GET 请求成功 |
| 201 | Created | POST 创建资源成功 |
| 204 | No Content | DELETE 删除成功,无返回体 |
| 400 | Bad Request | 参数校验失败 |
| 401 | Unauthorized | 未认证或认证失败 |
| 403 | Forbidden | 已认证但无权限 |
| 404 | Not Found | 资源不存在 |
| 422 | Unprocessable Entity | 请求体校验失败 |
| 500 | Internal 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 等。建议在继续学习之前,通过实际项目反复练习本章内容,确保对请求-响应链路的每个环节都有深刻理解。