⚠️ 重要:本文包含 API创建路径操作(Path Operations)查询参数与请求体响应模型与状态码错误处理(全局异常捕获与统一错误格式)

文中涉及的相关代码示例地址:https://github.com/m12305/hello-FastAPI

文章目录

1.1 Hello World — 第一个 FastAPI 应用

从这里开始,你正式进入 FastAPI 的世界。3 行代码启动一个 API,然后逐层揭开它的面纱。


1. 环境准备

# 确认 Python 版本 >= 3.8
python --version

# 创建虚拟环境(如果还没创建)
python -m venv venv

# 激活虚拟环境
# Windows:
venv\Scripts\activate
# Mac/Linux:
source venv/bin/activate

# 安装 FastAPI 和服务器
pip install fastapi uvicorn[standard]

验证安装:

pip show fastapi   # 应该显示版本号,如 0.115.x
pip show uvicorn   # 应该显示版本号

2. 最小应用:3 行代码

创建文件并运行你的第一个 FastAPI 应用:

# 文件名: main.py
from fastapi import FastAPI

app = FastAPI()                          # 创建应用实例


@app.get("/")                            # 注册路由:GET /
async def root():                        # 路径操作函数
    return {"message": "Hello World"}    # 自动转为 JSON

运行

uvicorn main:app --reload
#       │    │     │
#       │    │     └── --reload: 代码修改后自动重启(开发模式)
#       │    └──────── 应用实例名(变量名)
#       └───────────── 文件名(不含 .py)

打开浏览器访问:

🎉 你已经有了一个生产就绪的 API(自动文档、自动 JSON 序列化、高性能异步服务器)。


3. 发生了什么?逐行解析

from fastapi import FastAPI      # 导入 FastAPI 类

app = FastAPI()                  # 实例化一个 FastAPI 应用
                                 # app 是整个应用的核心,所有路由都注册在它上面
                                 # 参数说明(常用):
                                 #   title="My API"       → 文档标题
                                 #   description="..."    → 文档描述
                                 #   version="1.0.0"      → 版本号
                                 #   docs_url=None        → 禁用 Swagger
                                 #   redoc_url=None       → 禁用 ReDoc

@app.get("/")                    # 路径操作装饰器
                                 # @app.get     → 处理 GET 请求
                                 # @app.post    → 处理 POST 请求
                                 # @app.put     → 处理 PUT 请求
                                 # @app.delete  → 处理 DELETE 请求
                                 # "/"         → URL 路径

async def root():                # 路径操作函数(Path Operation Function)
    return {"message": "..."}    # 返回 dict → FastAPI 自动转为 JSON

FastAPI 应用初始化常用参数

app = FastAPI(
    title="我的第一个 API",
    description="这是一个学习 FastAPI 的示例项目",
    version="0.1.0",
    # docs_url="/api-docs",       # 自定义文档路径
    # openapi_url="/openapi.json", # 自定义 OpenAPI Schema 路径
)

4. Uvicorn 是什么?

浏览器/客户端 ──HTTP──> Uvicorn (ASGI 服务器) ──ASGI──> FastAPI 应用
                 <──          响应          <──        返回数据
组件 角色
Uvicorn ASGI 服务器 — 接收 HTTP 请求,管理进程/线程,转发给应用
FastAPI ASGI 应用 — 处理请求(路由、校验、序列化),返回数据
ASGI 异步服务器网关接口 — Python 异步 Web 服务器和应用之间的协议

FastAPI 不包含服务器,它需要 Uvicorn(或 Hypercorn)来接收网络请求。类比:Flask/Django 是 WSGI 应用 + Gunicorn。


5. --reload 开发模式

# 开发模式(代码修改自动重启)
uvicorn main:app --reload

# 等价于
uvicorn main:app --reload --reload-dir ./

# 生产模式(不要用 --reload!性能更好)
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
参数 说明 默认值
--host 绑定的 IP 127.0.0.1
--port 绑定的端口 8000
--reload 自动重启 关闭
--workers 工作进程数 1

6. 探索自动生成的文档

启动应用后,访问 http://127.0.0.1:8000/docs

你会看到:

  • 📋 你的路由 GET / 列在页面上
  • 🔽 展开后可以看到 Try it out 按钮
  • 点击后可以直接在浏览器中测试 API
  • 下面有 Response,展示了返回的 JSON
  • 整个页面遵循 OpenAPI 标准

这套文档是从你的代码自动推断的——你的函数参数类型、Pydantic 模型、docstring 都会自动反映在文档中。


7. 测试你的 API

方式 1:浏览器

直接访问 http://127.0.0.1:8000

方式 2:Swagger UI

访问 http://127.0.0.1:8000/docs → Try it out → Execute

方式 3:curl

curl http://127.0.0.1:8000/
# 输出: {"message":"Hello World"}

# 查看响应头
curl -i http://127.0.0.1:8000/
# HTTP/1.1 200 OK
# content-type: application/json
# ...

方式 4:Python httpx

import httpx

resp = httpx.get("http://127.0.0.1:8000/")
print(resp.status_code)  # 200
print(resp.json())       # {'message': 'Hello World'}

方式 5:FastAPI TestClient(后面详细学)

from fastapi.testclient import TestClient

client = TestClient(app)
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello World"}

8. 扩展你的第一个 API

# main.py
from fastapi import FastAPI

app = FastAPI(
    title="我的第一个 API",
    description="FastAPI 学习项目",
    version="0.1.0",
)


@app.get("/")
async def root():
    """根路径,返回欢迎信息"""
    return {"message": "Hello World", "version": "0.1.0"}


@app.get("/ping")
async def ping():
    """健康检查接口"""
    return {"status": "ok", "service": "my-api"}


@app.get("/time")
async def current_time():
    """返回服务器当前时间"""
    from datetime import datetime
    return {"current_time": datetime.now().isoformat()}


@app.get("/hello/{name}")
async def hello(name: str):
    """向指定用户问好"""
    return {"message": f"Hello, {name}!"}

保存后刷新 http://127.0.0.1:8000/docs,你会看到 4 个接口出现在文档中。


9. 常见问题排查

问题 1:端口被占用

# Error: [Errno 10048] address already in use
# 换一个端口
uvicorn main:app --reload --port 8001

问题 2:找不到模块

# ModuleNotFoundError: No module named 'main'
# 确认:
# 1. 你在正确的目录下运行命令
# 2. 文件名是 main.py
# 3. 文件名不含特殊字符

问题 3:没有自动重载

# 确保加了 --reload 参数
# 某些 IDE 保存时不会触发文件变更事件,手动 touch 一下

10. 实战练习

"""练习:创建一个个人简介 API

要求:
1. FastAPI 应用标题为 "个人简介 API"
2. 实现以下接口:
   - GET / → 返回 {"name": "你的名字", "role": "FastAPI Learner"}
   - GET /about → 返回你的学习目标和开始日期
   - GET /status → 返回 {"status": "online", "uptime": "计算从程序启动到现在的秒数"}
   
提示:uptime 可以在应用启动时记录 start_time = time.time()
"""

# 你的代码:

11. 本章检查清单

  • 成功安装了 fastapi 和 uvicorn
  • 创建并运行了第一个 FastAPI 应用
  • 在浏览器中看到了 {"message": "Hello World"}
  • 访问了 /docs 页面,并试用 Swagger UI
  • 访问了 /redoc 页面
  • 理解了 uvicorn main:app --reload 每个部分的含义
  • 理解了 FastAPI(应用)和 Uvicorn(服务器)的关系
  • 用 curl 或 httpx 测试了 API
  • 为应用添加了标题和描述
  • 完成了实战练习

🎯 恭喜!你已经有了一个可以运行的 API。接下来学路径操作,让 API 拥有真正的 CRUD 能力。

1.2 路径操作(Path Operations)

路径操作是 API 的骨架——URL 设计和 HTTP 方法的组合。掌握它就掌握了路由的完整控制权。


1. 什么是"路径操作"?

@app.get("/users/{user_id}")
\________/  \______________/
    │              │
    │              └── 路径(Path):URL 路径,可以包含动态参数
    └───────────────── 操作(Operation):HTTP 方法(GET/POST/PUT/DELETE/PATCH)

FastAPI 的路径操作 = HTTP 方法装饰器 + URL 路径 + 路径操作函数


2. 所有 HTTP 方法装饰器

from fastapi import FastAPI

app = FastAPI()

@app.get("/items")           # 读取资源列表
async def list_items():
    return [{"id": 1}, {"id": 2}]

@app.get("/items/{item_id}") # 读取单个资源(注意:有路径参数)
async def get_item(item_id: int):
    return {"id": item_id}

@app.post("/items")          # 创建资源
async def create_item():
    return {"message": "Item created"}

@app.put("/items/{item_id}") # 完整替换资源
async def replace_item(item_id: int):
    return {"message": f"Item {item_id} replaced"}

@app.patch("/items/{item_id}") # 部分更新资源
async def update_item(item_id: int):
    return {"message": f"Item {item_id} updated"}

@app.delete("/items/{item_id}") # 删除资源
async def delete_item(item_id: int):
    return {"message": f"Item {item_id} deleted"}

# 高级:自定义 HTTP 方法(或同时接受多个方法)
@app.api_route("/custom", methods=["GET", "POST", "PURGE"])
async def custom_handler():
    return {"message": "custom"}

3. 路径参数(Path Parameters)

路径参数是 URL 中动态的部分,用 {} 包裹:

@app.get("/users/{user_id}")
async def get_user(user_id: int):   # ← 函数参数名必须与 {user_id} 一致
    return {"user_id": user_id}

# GET /users/42    → {"user_id": 42}
# GET /users/abc   → 自动返回 422 错误(类型不匹配,期望 int)

类型转换

FastAPI 根据函数参数的类型标注自动转换:

@app.get("/items/{item_id}")
async def get_item(item_id: int):    # ← 自动转为 int
    return {"item_id": item_id, "type": type(item_id).__name__}
# GET /items/123 → {"item_id": 123, "type": "int"}

@app.get("/price/{amount}")
async def get_price(amount: float):  # ← 自动转为 float
    ...
# GET /price/19.99 → amount = 19.99

@app.get("/flag/{value}")
async def get_flag(value: bool):     # ← 自动转为 bool
    ...
# GET /flag/true   → value = True
# GET /flag/1      → value = True
# GET /flag/false  → value = False

4. 路径参数的顺序与路由优先级

# ⚠️ 重要:固定路径必须写在动态路径前面!

# ✅ 正确顺序
@app.get("/users/me")                # 先注册固定路径
async def get_current_user():
    return {"username": "Alice"}

@app.get("/users/{user_id}")         # 再注册动态路径
async def get_user(user_id: int):
    return {"user_id": user_id}

# ❌ 错误顺序:/users/{user_id} 会吞掉 /users/me
# GET /users/me → user_id="me" → 类型转换 int("me") → 报错

5. 路径参数的枚举约束

当路径参数的取值只有几种可能时,用枚举限制:

from enum import Enum

class Category(str, Enum):
    electronics = "electronics"
    clothing = "clothing"
    food = "food"

@app.get("/products/{category}")
async def get_products(category: Category):
    """category 只能是 electronics, clothing, food"""
    return {"category": category, "products": [...]}

# GET /products/electronics  → ✅
# GET /products/books        → ❌ 422 错误,自动提示可选值

枚举的好处:

  • 类型安全 — 非法值自动拒绝
  • 文档自描述 — Swagger 中自动显示可选值作为下拉菜单

6. 定义多个路径参数的场景

# 层级嵌套:一个用户的多篇文章
@app.get("/users/{user_id}/posts/{post_id}")
async def get_user_post(user_id: int, post_id: int):
    return {"user_id": user_id, "post_id": post_id}

# GET /users/42/posts/5 → {"user_id": 42, "post_id": 5}

# 更复杂的层级
@app.get("/orgs/{org_id}/teams/{team_id}/members/{member_id}")
async def get_team_member(org_id: int, team_id: int, member_id: int):
    return {
        "org": org_id,
        "team": team_id,
        "member": member_id,
    }

7. @app.api_route() 高级用法

当你需要一个路由同时响应多种 HTTP 方法时:

@app.api_route("/webhook", methods=["GET", "POST"])
async def webhook():
    """同一个处理函数同时处理 GET 和 POST"""
    return {"status": "received"}

# 或者对请求对象做精细控制
from fastapi import Request

@app.api_route("/dynamic", methods=["GET", "POST", "PUT", "DELETE"])
async def dynamic_handler(request: Request):
    """根据 request.method 做不同处理"""
    method = request.method
    if method == "GET":
        return {"action": "read"}
    elif method == "POST":
        return {"action": "create"}
    elif method == "PUT":
        return {"action": "replace"}
    else:
        return {"action": "delete"}

8. 路径操作装饰器的额外参数

@app.get(
    "/items/{item_id}",
    response_model=ItemResponse,    # 响应模型(第 1.4 章详讲)
    status_code=200,                # 默认成功状态码
    tags=["items"],                 # Swagger 文档中的分组标签
    summary="获取单个商品",          # Swagger 中的简短摘要
    description="通过商品 ID 获取商品的详细信息",  # 详细描述
    deprecated=True,                # 标记为已废弃(文档中会显示)
)
async def get_item(item_id: int):
    ...

这些参数影响的是文档展示响应行为,不影响路由匹配。


9. 实战:用临时字典模拟数据库

from fastapi import FastAPI
from enum import Enum

app = FastAPI(title="TODO API")

# 用字典当临时数据库
todos: dict[int, dict] = {}
next_id = 1

@app.get("/todos")
async def list_todos():
    """获取所有 Todo"""
    return list(todos.values())

@app.get("/todos/{todo_id}")
async def get_todo(todo_id: int):
    """获取单个 Todo"""
    if todo_id not in todos:
        from Fastapi import HTTPException
        raise HTTPException(status_code=404, detail="Todo not found")
    return todos[todo_id]

@app.post("/todos")
async def create_todo(title: str, done: bool = False):
    """创建新的 Todo"""
    global next_id
    todos[next_id] = {"id": next_id, "title": title, "done": done}
    next_id += 1

@app.put("/todos/{todo_id}")
async def update_todo(todo_id: int, title: str, done: bool):
    """完整更新 Todo"""
    if todo_id not in todos:
        from Fastapi import HTTPException
        raise HTTPException(status_code=404, detail="Todo not found")
    ...

@app.delete("/todos/{todo_id}")
async def delete_todo(todo_id: int):
    """删除 Todo"""
    if todo_id not in todos:
        from Fastapi import HTTPException
        raise HTTPException(status_code=404, detail="Todo not found")
    ...

注意: 上面用函数参数 title: str 直接接收数据,这只是临时写法。下一章(1.3)会学到正确的请求体处理方式。


10. 实践练习

"""练习:书籍管理 API 骨架

要求:
1. 定义 BookCategory 枚举(fiction, non-fiction, science, history)
2. 实现以下路由:
   - GET /books → 返回书籍列表(用硬编码列表)
   - GET /books/{book_id} → 返回单本书(不存在返回 404)
   - GET /books/category/{category} → 按分类筛选
   - POST /books → 创建书籍(用临时字典存储)
   - DELETE /books/{book_id} → 删除书籍

3. 给每个路由添加 tags、summary 参数
4. 访问 /docs 确认文档展示正常
"""

# 你的代码:

11. 本章检查清单

  • 理解路径操作 = HTTP 方法 + URL 路径 + 处理函数
  • 能使用全部 6 种装饰器(get, post, put, patch, delete, api_route)
  • 理解路径参数 {param} 的语法和类型自动转换
  • 知道固定路径要写在动态路径前面
  • 能用枚举约束路径参数的取值
  • 能设计嵌套的 URL 层级结构
  • 了解装饰器的 tags、summary、deprecated 等文档参数
  • 完成实战练习

🎯 路径操作是 API 的入口。下一章学习查询参数和请求体,让 API 能真正处理用户提交的数据。

1.3 查询参数与请求体

API 的核心价值是接收和处理数据。这一章你将学会接收查询参数和 JSON 请求体的正确姿势。


1. 三种数据来源总览

POST /users/42?lang=zh&notify=true
\________/ \___________/ \____________________/
    │           │                 │
 请求体    路径参数          查询参数
 (JSON)   (URL 一部分)     (?key=value)
数据来源 在 URL 中 必选/可选 典型用途
路径参数 /users/{id} 必选 资源唯一标识
查询参数 ?page=1&size=20 可选(有默认值) 过滤、排序、分页
请求体 HTTP Body(JSON) 看 API 设计 创建/更新时的数据

2. 查询参数(Query Parameters)

基础用法

from fastapi import FastAPI

app = FastAPI()

# 函数参数中不在路径中的 → 自动成为查询参数
@app.get("/users")
async def list_users(
    page: int = 1,        # 有默认值 → 可选参数,?page=2
    size: int = 20,       # 有默认值 → 可选参数,?size=50
    active_only: bool = False,
):
    return {
        "page": page,
        "size": size,
        "active_only": active_only,
    }

# GET /users                              → 使用所有默认值
# GET /users?page=2                       → page=2, 其余默认
# GET /users?page=2&size=50&active_only=true → 全部指定

必选 vs 可选

@app.get("/search")
async def search(
    q: str,                    # ← 没有默认值 → 必选!
    limit: int = 10,           # ← 有默认值 → 可选
):
    return {"q": q, "limit": limit}

# GET /search          → 422 错误(缺少必选参数 q)
# GET /search?q=python → ✅ {"q": "python", "limit": 10}

可选参数用 Optional

from typing import Optional

@app.get("/users")
async def list_users(
    status: Optional[str] = None,  # 可以为 None,不传就是 None
    tag: str | None = None,        # Python 3.10+ 写法
):
    filters = {}
    if status:
        filters["status"] = status
    if tag:
        filters["tag"] = tag
    return {"filters": filters}

3. 用 Query() 添加高级校验

当需要给查询参数添加校验规则和元数据时,用 Query()

from fastapi import FastAPI, Query
from typing import Annotated

app = FastAPI()

@app.get("/items")
async def list_items(
    # Annotated 写法(FastAPI 推荐)
    page: Annotated[int, Query(ge=1, description="页码")] = 1,
    size: Annotated[int, Query(ge=1, le=100, description="每页数量")] = 20,
    # 等价传统写法(不推荐,类型标注和校验混在一起)
    # page: int = Query(1, ge=1, description="页码"),
):
    return {"page": page, "size": size}

# GET /items?page=0    → 422 错误(page 必须 >= 1)
# GET /items?size=200  → 422 错误(size 必须 <= 100)

Query() 常用参数

@app.get("/search")
async def search_items(
    q: Annotated[
        str,
        Query(
            min_length=1,              # 最小长度
            max_length=50,             # 最大长度
            pattern=r"^[a-zA-Z0-9 ]+$", # 正则匹配
            title="搜索关键词",
            description="搜索商品名称或描述",
            alias="query",            # URL 中使用 ?query=xxx 而不是 ?q=xxx
            deprecated=True,          # 在文档中标记为废弃
            include_in_schema=False,  # 从 OpenAPI 文档中隐藏
        ),
    ] = None,
):
    return {"q": q}

4. 请求体(Request Body)

请求体是 POST/PUT/PATCH 时通过 HTTP Body 发送的数据,通常为 JSON 格式。

用 Pydantic 模型定义请求体

from pydantic import BaseModel, Field

class ItemCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    price: float = Field(gt=0)
    description: str | None = Field(default=None, max_length=1000)
    tax: float | None = None

@app.post("/items")
async def create_item(item: ItemCreate):  # ← Pydantic 模型 = 请求体
    # item 已经是校验过的 ItemCreate 实例
    return {
        "name": item.name,
        "price": item.price,
        "price_with_tax": item.price + (item.tax or 0),
    }

测试请求体

curl -X POST http://127.0.0.1:8000/items \
  -H "Content-Type: application/json" \
  -d '{"name": "Widget", "price": 19.99, "tax": 1.5}'

或在 Swagger UI (/docs) 中直接填入 JSON。


5. 路径参数 + 查询参数 + 请求体 组合

from pydantic import BaseModel

class OrderCreate(BaseModel):
    product_id: int
    quantity: int = Field(ge=1)
    address: str

@app.post("/users/{user_id}/orders")
async def create_order(
    user_id: int,                          # 路径参数
    order: OrderCreate,                    # 请求体
    send_notification: bool = False,       # 查询参数
    priority: str = "normal",             # 查询参数
):
    return {
        "message": f"Order created for user {user_id}",
        "product_id": order.product_id,
        "quantity": order.quantity,
        "address": order.address,
        "notification": send_notification,
        "priority": priority,
    }

# POST /users/42/orders?send_notification=true&priority=high
# Body: {"product_id": 101, "quantity": 3, "address": "Beijing"}
#
# FastAPI 自动把数据分配到正确的参数:
# - user_id=42       → 来自 URL 路径
# - order            → 来自请求体 JSON
# - send_notification → 来自查询参数
# - priority         → 来自查询参数

6. 多请求体参数

当一个接口需要接收多个逻辑上独立的数据块时:

class UserInfo(BaseModel):
    name: str
    email: str

class ShippingAddress(BaseModel):
    street: str
    city: str
    zip_code: str

@app.post("/checkout")
async def checkout(
    user: UserInfo,          # 第一个 JSON 对象 {"name":..., "email":...}
    address: ShippingAddress,# 第二个 JSON 对象 {"street":..., ...}
):
    return {
        "user": user,
        "shipping": address,
    }

# 请求体需要这样发送:
# {
#   "user": {"name": "Alice", "email": "alice@example.com"},
#   "address": {"street": "123 Main St", "city": "Beijing", "zip_code": "100000"}
# }

用 Body(embed=True) 嵌入单个字段

from fastapi import Body

@app.post("/users")
async def create_user(
    name: str = Body(...),     # ... 表示必填
    email: str = Body(...),
    age: int = Body(25),       # 可选 Body 参数
):
    return {"name": name, "email": email, "age": age}

# 外部调用时需要包裹一层:
# {"name": "Alice", "email": "alice@example.com", "age": 25}

7. 数据来源的优先级与识别规则

FastAPI 的参数识别规则(按优先级):

@app.get("/items/{item_id}")
async def some_handler(
    item_id: int,              # 1. 在路径中 → 路径参数
    query_param: str = "xxx",  # 2. 基本类型 + 不在路径中 → 查询参数
    body_param: MyModel,       # 3. Pydantic 模型 → 请求体
    special: int = Body(...),  # 4. 显式用了 Body() → 请求体
    header: str = Header(...), # 5. 显式用了 Header() → 请求头
    cookie: str = Cookie(...), # 6. 显式用了 Cookie() → Cookie
):
    ...

8. 实战:构建一个完整的商品搜索 API

from fastapi import FastAPI, Query, Body
from pydantic import BaseModel, Field
from typing import Annotated, Optional
from enum import Enum

app = FastAPI(title="商品管理 API")

# 模拟数据库
products_db: list[dict] = [
    {"id": 1, "name": "机械键盘", "category": "electronics", "price": 399.0, "in_stock": True},
    {"id": 2, "name": "Python 编程书", "category": "books", "price": 79.0, "in_stock": True},
    {"id": 3, "name": "T恤", "category": "clothing", "price": 129.0, "in_stock": False},
]

class ProductCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    category: str
    price: float = Field(gt=0)
    in_stock: bool = True

# GET /products —— 搜索 + 筛选 + 分页
@app.get("/products")
async def search_products(
    q: Annotated[Optional[str], Query(description="搜索关键词")] = None,
    category: Annotated[Optional[str], Query(description="分类筛选")] = None,
    min_price: Annotated[Optional[float], Query(ge=0, description="最低价")] = None,
    max_price: Annotated[Optional[float], Query(ge=0, description="最高价")] = None,
    page: Annotated[int, Query(ge=1)] = 1,
    size: Annotated[int, Query(ge=1, le=50)] = 10,
):
    results = products_db

    if q:
        results = [p for p in results if q.lower() in p["name"].lower()]
    if category:
        results = [p for p in results if p["category"] == category]
    if min_price is not None:
        results = [p for p in results if p["price"] >= min_price]
    if max_price is not None:
        results = [p for p in results if p["price"] <= max_price]

    # 分页
    start = (page - 1) * size
    end = start + size
    return {
        "total": len(results),
        "page": page,
        "size": size,
        "items": results[start:end],
    }

# POST /products —— 创建商品
@app.post("/products", status_code=201)
async def create_product(product: ProductCreate):
    new_id = max(p["id"] for p in products_db) + 1
    new_product = {"id": new_id, **product.model_dump()}
    products_db.append(new_product)
    return new_product

9. 实战练习

"""练习:图书管理 API

实现以下接口:
1. POST /books —— 创建图书
   请求体包含:title(必填), author(必填), year(必填,1900-2026), 
               pages(可选,>0), isbn(可选,13位数字)
   
2. GET /books —— 搜索图书
   查询参数:author(可选), min_year(可选), max_year(可选), 
             keyword(可选,搜索 title), 
             page(>=1), size(1-100)

要求:
- 使用 Pydantic 模型定义请求体
- 使用 Annotated[..., Query(...)] 添加校验
- 用内存列表存储数据
"""

# 你的代码:

10. 本章检查清单

  • 理解三种数据来源:路径参数、查询参数、请求体的区别
  • 能区分必选参数和可选参数
  • 能用 Query() 给查询参数添加校验规则
  • 能用 Pydantic 模型定义请求体
  • 能在同一个路由中混合使用三种数据来源
  • 理解 FastAPI 的参数识别规则
  • 能实现带搜索、筛选、分页的列表接口
  • 完成实战练习

🎯 数据输入掌握了。下一章学响应模型和状态码,掌控 API 的输出。

1.4 响应模型与状态码

控制 API 的输出和输入同样重要。这一章教你精确控制返回什么数据、以什么格式、带什么状态码。


1. 默认行为

from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"id": user_id, "name": "Alice", "password_hash": "abc123"}

FastAPI 默认:

  • 把你 return 的任何东西序列化为 JSON
  • 状态码默认 200(GET/PUT/PATCH)或 201(POST)
  • 不限制返回的字段

2. response_model —— 控制响应结构

response_model 是 FastAPI 最重要的响应控制手段:

from pydantic import BaseModel

# 数据库中的用户(包含敏感字段)
fake_db = {
    1: {"id": 1, "name": "Alice", "email": "alice@example.com", 
        "password_hash": "$2b$12$...", "internal_notes": "VIP customer"}
}

# 公开的响应模型(只暴露安全字段)
class UserResponse(BaseModel):
    id: int
    name: str
    email: str

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    user = fake_db.get(user_id)
    # user 包含 password_hash 和 internal_notes
    return user
    # FastAPI 自动过滤:只返回 UserResponse 中定义的字段!
    # {"id": 1, "name": "Alice", "email": "alice@example.com"}

为什么用 response_model?

  1. 安全 — 绝不泄露密码哈希、内部备注等敏感数据
  2. 文档 — Swagger 自动展示响应结构
  3. 校验 — 确保返回的数据符合 Schema
  4. 性能 — 过滤掉不需要返回的大字段

3. 响应字段的过滤

response_model_include / exclude

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    is_active: bool
    created_at: str

# 只看部分字段
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user_basic(user_id: int):
    ...

# 只包含指定字段
@app.get("/users/{user_id}", response_model=UserResponse,
         response_model_include={"id", "name"})
async def get_user_minimal(user_id: int):
    ...
    # 返回: {"id": 1, "name": "Alice"}

# 排除指定字段
@app.get("/users/{user_id}", response_model=UserResponse,
         response_model_exclude={"created_at", "is_active"})
async def get_user_clean(user_id: int):
    ...
    # 返回: {"id": 1, "name": "Alice", "email": "alice@example.com"}

嵌套列表的响应模型

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

# 返回列表时,用 List[ItemResponse]
from typing import List

@app.get("/items", response_model=List[ItemResponse])
async def list_items():
    return [
        {"id": 1, "name": "Widget", "price": 19.99, "_internal": "xxx"},
        {"id": 2, "name": "Gadget", "price": 29.99, "_internal": "yyy"},
    ]
    # 每个元素都会被 ItemResponse 过滤

4. 状态码(Status Code)

设置默认状态码

from fastapi import FastAPI, status

@app.post("/users", status_code=201)                        # 数字
async def create_user():
    return {"message": "created"}

@app.post("/users", status_code=status.HTTP_201_CREATED)    # 推荐:用 status 常量
async def create_user():
    return {"message": "created"}

# 常用 status 常量:
# status.HTTP_200_OK
# status.HTTP_201_CREATED
# status.HTTP_204_NO_CONTENT
# status.HTTP_400_BAD_REQUEST
# status.HTTP_401_UNAUTHORIZED
# status.HTTP_403_FORBIDDEN
# status.HTTP_404_NOT_FOUND
# status.HTTP_422_UNPROCESSABLE_ENTITY
# status.HTTP_500_INTERNAL_SERVER_ERROR

动态设置状态码

from fastapi import Response

@app.get("/items/{item_id}")
async def get_item(item_id: int, response: Response):
    if item_id == 99:
        response.status_code = 302           # 临时重定向
        response.headers["Location"] = "/items/100"
        return None
    return {"id": item_id}

@app.post("/items")
async def create_item(item: ItemCreate, response: Response):
    # 创建成功,手动设置 201 + Location 头
    response.status_code = status.HTTP_201_CREATED
    response.headers["Location"] = f"/items/{new_id}"
    return new_item

5. 响应类型大全

除了默认的 JSON 响应,FastAPI 支持多种响应类型:

from fastapi import FastAPI
from fastapi.responses import (
    JSONResponse,
    HTMLResponse,
    PlainTextResponse,
    FileResponse,
    StreamingResponse,
    RedirectResponse,
)

app = FastAPI()

# 1. JSON 响应(默认,也可以显式使用)
@app.get("/json")
async def json_response():
    return JSONResponse(
        content={"message": "hello"},
        status_code=200,
        headers={"X-Custom-Header": "value"},
    )

# 2. HTML 响应(配合 Jinja2 模板)
@app.get("/html", response_class=HTMLResponse)
async def html_response():
    return """
    <html>
        <head><title>FastAPI</title></head>
        <body><h1>Hello FastAPI!</h1></body>
    </html>
    """

# 3. 纯文本
@app.get("/text", response_class=PlainTextResponse)
async def text_response():
    return "Hello, this is plain text"

# 4. 文件下载
@app.get("/download")
async def download_file():
    return FileResponse(
        path="path/to/file.pdf",
        filename="报告.pdf",              # 下载时的文件名
        media_type="application/pdf",
    )

# 5. 流式响应(大文件、实时数据)
@app.get("/stream")
async def stream_data():
    async def generate():
        for i in range(10):
            yield f"data: chunk {i}\n\n"
    return StreamingResponse(generate(), media_type="text/plain")

# 6. 重定向
@app.get("/old-path")
async def old_path():
    return RedirectResponse(url="/new-path", status_code=301)

6. 响应的最佳实践:请求/响应分离

from pydantic import BaseModel, Field
from datetime import datetime

# ===== 输入模型 =====
class UserCreate(BaseModel):
    name: str = Field(min_length=2)
    email: str
    password: str = Field(min_length=8)

# ===== 输出模型 =====
class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    created_at: str

# ===== 数据库模型(SQLAlchemy 前先用 dataclass 模拟) =====
class UserInDB(BaseModel):
    """内部模型,包含所有字段"""
    id: int
    name: str
    email: str
    password_hash: str
    created_at: str = datetime.now().isoformat()

# ===== 路由 =====
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user_in: UserCreate):
    """创建用户:接收 UserCreate,返回 UserResponse(无密码!)"""
    # 模拟:创建数据库记录
    db_user = UserInDB(
        id=1,
        name=user_in.name,
        email=user_in.email,
        password_hash=f"hashed_{user_in.password}",
    )
    return db_user  # response_model 自动过滤掉 password_hash

黄金法则: 永远不要把数据库模型直接返回给客户端。用专门的响应模型过滤敏感字段。


7. response_model 与类型安全

response_model 还会在返回时校验数据

class UserResponse(BaseModel):
    id: int
    name: str

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    # 假设某天数据库返回了错误类型
    return {"id": "not-a-number", "name": "Alice"}
    # ❌ FastAPI 会抛异常,因为 id 应该是 int 而不是 str
    # 这帮你发现数据层的 bug!

8. Response 对象 vs response_model

方式 校验 过滤 文档 适用场景
response_model=PydanticModel 99% 的 REST API
Response(content=...) 需要完全控制响应时
JSONResponse(content=...) 自定义 JSON 响应头
return dict 快速原型

9. 实战练习

"""练习:用户管理 API 的响应设计

1. 定义以下模型:
   - UserCreate(输入):name, email, password(>=8位)
   - UserResponse(公开输出):id, name, email, is_active, created_at
   - UserDetailResponse(详情输出):包含 UserResponse + 最近登录时间

2. 实现接口:
   - POST /users → 201, response_model=UserResponse
   - GET /users → 200, response_model=List[UserResponse]
   - GET /users/{id} → 200, response_model=UserDetailResponse
   - DELETE /users/{id} → 204 (No Content)

3. 要求:
   - 密码绝不返回
   - 创建时返回 Location 头
   - 删除成功返回 204 无内容
"""

# 你的代码:

10. 本章检查清单

  • 理解 response_model 的四大作用:安全、文档、校验、性能
  • 会用 response_model_include / exclude 过滤字段
  • 能用 status_codestatus.HTTP_* 常量设置状态码
  • 了解 6 种常用响应类型及其使用场景
  • 理解请求模型和响应模型分离的设计原则
  • 知道如何返回列表的响应模型
  • 知道 response_model 也会校验返回数据
  • 完成实战练习

🎯 输入和输出都掌握了。下一章学错误处理,让你的 API 健壮而专业。

1.5 错误处理

一个好的 API 不仅能在正常时工作,更要在出错时给出清晰、专业的反馈。这一章教你掌控所有异常场景。


1. HTTPException —— 主动抛出 HTTP 错误

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

fake_items = {1: {"id": 1, "name": "Widget"}}

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id not in fake_items:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,  # 状态码
            detail=f"Item {item_id} not found",       # 错误描述(必填)
            headers={"X-Error-Code": "ITEM_NOT_FOUND"}, # 可选的额外响应头
        )
    return fake_items[item_id]

# GET /items/999
# → 404
# → {"detail": "Item 999 not found"}
# → 响应头: x-error-code: ITEM_NOT_FOUND

HTTPException 参数

参数 类型 说明
status_code int HTTP 状态码
detail str / dict / list 错误详情(可以是 JSON 可序列化的任意类型)
headers dict 额外的响应头

2. 不同的错误,不同的状态码

fake_users = {1: {"id": 1, "name": "Alice", "role": "user"}}

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # 404 — 资源不存在
    if user_id not in fake_users:
        raise HTTPException(status_code=404, detail="用户不存在")

    return fake_users[user_id]

@app.patch("/users/{user_id}/role")
async def update_role(user_id: int, new_role: str):
    # 404
    if user_id not in fake_users:
        raise HTTPException(status_code=404, detail="用户不存在")

    # 422 — 参数语义错误
    allowed_roles = {"admin", "editor", "viewer"}
    if new_role not in allowed_roles:
        raise HTTPException(
            status_code=422,
            detail=f"无效的角色:{new_role}。可选:{allowed_roles}"
        )

    fake_users[user_id]["role"] = new_role
    return fake_users[user_id]

@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
    # 409 — 冲突
    if fake_users[user_id]["role"] == "admin":
        raise HTTPException(status_code=409, detail="不能删除管理员用户")

    del fake_users[user_id]
    return {"message": "删除成功"}

场景 → 状态码速查

场景 状态码
资源不存在 404 Not Found
参数校验失败 422 Unprocessable Entity
未登录/Token 无效 401 Unauthorized
已登录但权限不足 403 Forbidden
业务冲突(重复操作等) 409 Conflict
请求过于频繁 429 Too Many Requests
服务不可用 503 Service Unavailable

3. 自定义异常处理器

你可以全局接管特定异常的处理方式:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

# ===== 1. 自定义异常类 =====
class BusinessError(Exception):
    def __init__(self, message: str, error_code: str = "BUSINESS_ERROR"):
        self.message = message
        self.error_code = error_code

class ItemNotFoundError(BusinessError):
    def __init__(self, item_id: int):
        super().__init__(
            message=f"商品 {item_id} 不存在",
            error_code="ITEM_NOT_FOUND"
        )

# ===== 2. 注册全局异常处理器 =====
@app.exception_handler(BusinessError)
async def business_error_handler(request: Request, exc: BusinessError):
    return JSONResponse(
        status_code=400,
        content={
            "error": exc.error_code,
            "message": exc.message,
        },
    )

# ===== 3. 在路由中使用 =====
@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id not in fake_items:
        raise ItemNotFoundError(item_id)  # 自动被上方处理器捕获
    return fake_items[item_id]

4. 覆盖 FastAPI 内置的异常处理器

from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """自定义参数校验错误的返回格式"""
    # 提取所有校验错误
    errors = []
    for error in exc.errors():
        errors.append({
            "field": " -> ".join(str(loc) for loc in error["loc"]),
            "message": error["msg"],
            "type": error["type"],
        })

    return JSONResponse(
        status_code=422,
        content={
            "error": "VALIDATION_ERROR",
            "message": "请求参数校验失败",
            "details": errors,
        },
    )

# 现在所有参数校验错误都会返回这种格式,而不是默认格式

默认格式 vs 自定义格式

// FastAPI 默认格式:
{"detail": [{"loc": ["body", "name"], "msg": "field required", "type": "missing"}]}

// 自定义后:
{
  "error": "VALIDATION_ERROR",
  "message": "请求参数校验失败",
  "details": [
    {"field": "body -> name", "message": "field required", "type": "missing"}
  ]
}

5. 设计统一的错误响应格式

企业级 API 通常有统一的错误响应结构:

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
import traceback

app = FastAPI()

# ===== 统一错误响应模型 =====
def error_response(
    status_code: int,
    error_type: str,
    message: str,
    details: list | None = None,
    request_id: str | None = None,
) -> dict:
    return {
        "success": False,
        "error": {
            "type": error_type,
            "code": status_code,
            "message": message,
            "details": details or [],
            "request_id": request_id,
        },
    }

# ===== 处理 HTTPException =====
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content=error_response(
            status_code=exc.status_code,
            error_type="HTTP_ERROR",
            message=str(exc.detail),
        ),
    )

# ===== 处理参数校验错误 =====
@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    details = [
        {"field": " -> ".join(str(l) for l in e["loc"]), "message": e["msg"]}
        for e in exc.errors()
    ]
    return JSONResponse(
        status_code=422,
        content=error_response(
            status_code=422,
            error_type="VALIDATION_ERROR",
            message="请求参数校验失败",
            details=details,
        ),
    )

# ===== 兜底:捕获所有未处理异常 =====
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    # 生产环境中,这里应该记录日志、发送告警
    print("UNHANDLED ERROR:", traceback.format_exc())
    return JSONResponse(
        status_code=500,
        content=error_response(
            status_code=500,
            error_type="INTERNAL_ERROR",
            message="服务器内部错误,请稍后重试",
        ),
    )

6. 错误的 HTTPException 使用方式

# ✅ 好:清晰的错误信息
raise HTTPException(
    status_code=404,
    detail="Task with id 42 not found"
)

# ✅ 好:detail 可以是 dict(给前端更多信息)
raise HTTPException(
    status_code=409,
    detail={
        "message": "用户名已存在",
        "suggestion": "请更换用户名或尝试登录",
        "field": "username",
    }
)

# ❌ 坏:信息太模糊
raise HTTPException(status_code=400, detail="Error")

# ❌ 坏:泄露内部信息
raise HTTPException(status_code=500, detail=traceback.format_exc())

# ✅ 好:生产环境只返回通用信息,内部错误记录到日志
raise HTTPException(status_code=500, detail="服务器内部错误")

7. 实战:完整的错误处理方案

from fastapi import FastAPI, HTTPException, Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from pydantic import BaseModel, Field, field_validator

app = FastAPI(title="错误处理示例")

# ---------- 数据 ----------
users_db = {
    1: {"id": 1, "username": "alice", "email": "alice@example.com", "is_active": True},
}

# ---------- 模型 ----------
class UserCreate(BaseModel):
    username: str = Field(min_length=3, max_length=20)
    email: str
    age: int = Field(ge=18, le=120)

    @field_validator("username")
    @classmethod
    def username_no_spaces(cls, v: str) -> str:
        if " " in v:
            raise ValueError("用户名不能包含空格")
        return v

# ---------- 全局异常处理器 ----------
@app.exception_handler(HTTPException)
async def custom_http_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "success": False,
            "error": str(exc.detail),
            "status_code": exc.status_code,
        },
    )

@app.exception_handler(Exception)
async def global_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={
            "success": False,
            "error": "服务器内部错误",
            "status_code": 500,
        },
    )

# ---------- 路由 ----------
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    """获取用户:404 示例"""
    if user_id not in users_db:
        raise HTTPException(
            status_code=404,
            detail=f"用户 {user_id} 不存在"
        )
    return {"success": True, "data": users_db[user_id]}

@app.post("/users", status_code=201)
async def create_user(user: UserCreate):
    """创建用户:409 冲突示例"""
    # 检查用户名重复
    for u in users_db.values():
        if u["username"] == user.username:
            raise HTTPException(
                status_code=409,
                detail={"message": "用户名已存在", "field": "username"}
            )
    new_id = max(users_db) + 1
    users_db[new_id] = {"id": new_id, **user.model_dump(), "is_active": True}
    return {"success": True, "data": users_db[new_id]}

@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
    """删除用户:403 示例"""
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail=f"用户 {user_id} 不存在")
    if user_id == 1:
        raise HTTPException(status_code=403, detail="不能删除超级管理员")
    del users_db[user_id]
    return {"success": True, "message": "删除成功"}

8. 实战练习

"""练习:库存管理 API 的错误处理

实现以下接口,每个接口都要有对应的错误处理:

1. GET /products/{product_id}
   - 不存在 → 404
   
2. POST /products
   - name 和 price 必填(依赖 Pydantic 自动校验)
   - name 重复 → 409
   
3. POST /orders
   - product_id 不存在 → 404
   - 库存不足 → 409(附上当前库存量)
   - quantity <= 0 → 422
   
4. 添加一个全局异常处理器,统一错误响应格式为:
   {"error": true, "code": 状态码, "message": "错误信息"}

5. 添加一个自定义的 InsufficientStockError 异常类和对应的处理器
"""

# 你的代码:

9. 本章检查清单

  • 能用 HTTPException 主动抛出 HTTP 错误
  • 能为不同场景选择正确的 HTTP 状态码
  • 能自定义异常处理器 @app.exception_handler()
  • 能覆盖 FastAPI 内置的校验错误处理器
  • 能为 API 设计统一的错误响应格式
  • 理解生产环境中不要泄露内部错误信息的原则
  • 能用 detail 传递结构化的错误信息
  • 完成实战练习

10. 阶段 1 总结

你已经学完了 FastAPI 核心的 5 个章节:

章节 核心能力
1.1 Hello World 创建应用、运行服务器、访问文档
1.2 路径操作 定义路由、路径参数、HTTP 方法
1.3 查询参数与请求体 接收数据、Pydantic 校验、Query() 参数
1.4 响应模型与状态码 控制输出、过滤敏感字段、设置状态码
1.5 错误处理 抛出异常、统一错误格式、全局异常处理器

你现在可以: 构建一个完整的、带数据校验、有专业错误处理的 CRUD API。


🎯 阶段 1 完成! 回顾所有章节的检查清单,确认全部勾选后,进入阶段 2:请求与响应,学习依赖注入、中间件、文件上传等进阶主题。

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐