⚠️ 重要：如果你已经熟悉 Python 类型标注、async/await 和 HTTP 协议，可以快速过一遍，直接进入阶段 1。

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

文章目录

0.1 Python 类型标注（Type Hints）

FastAPI 的核心机制 —— 自动参数解析、数据验证、文档生成 —— 全部建立在 Python 类型标注之上。这一章必须学扎实。

1. 为什么需要类型标注？

Python 是动态类型语言，变量可以随时改变类型：

x = 1        # int
x = "hello"  # str  完全合法，但容易埋下 bug

类型标注让你声明意图，让工具帮你提前发现错误：

def add(a: int, b: int) -> int:
    return a + b

add(1, 2)       # ✅ 正确
add(1, "hello")  # ⚠️ IDE 会标红警告，但 Python 解释器仍然能运行

关键认知： Python 的类型标注不影响运行时，它是给人类和工具（IDE、mypy、FastAPI）看的元数据。

2. 基本类型标注

# 基本类型
name: str = "Alice"
age: int = 25
price: float = 19.99
is_active: bool = True

# 函数参数和返回值
def greet(name: str) -> str:
    return f"Hello, {name}!"

def divide(a: float, b: float) -> float:
    if b == 0:
        raise ValueError("除数不能为零")
    return a / b

3. 容器类型标注

from typing import List, Dict, Tuple, Set, Optional

# 列表：元素都是同一类型
users: List[str] = ["Alice", "Bob", "Charlie"]
scores: List[int] = [85, 92, 78]

# 字典：指定 key 和 value 的类型
user_age: Dict[str, int] = {"Alice": 25, "Bob": 30}

# 元组：固定长度，每个位置有特定类型
user_info: Tuple[str, int, str] = ("Alice", 25, "alice@example.com")

# 集合
tags: Set[str] = {"python", "fastapi", "web"}

# Optional = 可以是某类型，也可以是 None
def find_user(name: str) -> Optional[str]:
    """返回用户邮箱，找不到返回 None"""
    users = {"Alice": "alice@example.com"}
    return users.get(name)  # 可能返回 None

Python 3.10+ 新语法（推荐）

# 如果你使用的是 Python 3.10+，可以用内置小写类型
users: list[str] = ["Alice", "Bob"]
scores: list[int] = [85, 92]
user_age: dict[str, int] = {"Alice": 25}
user_info: tuple[str, int, str] = ("Alice", 25, "alice@example.com")
tags: set[str] = {"python", "fastapi"}

# X | None 替代 Optional[X]
def find_user(name: str) -> str | None:
    users = {"Alice": "alice@example.com"}
    return users.get(name)

4. Union 与 Any

from typing import Union, Any

# Union：可以是多种类型之一
def process(data: Union[int, str]) -> str:
    if isinstance(data, int):
        return str(data * 2)
    return data.upper()

# Python 3.10+ 简写
def process(data: int | str) -> str:
    ...

# Any：任意类型 —— 尽量少用，用了它就等于放弃了类型检查
def parse_json(raw: str) -> Any:
    import json
    return json.loads(raw)

5. Literal 类型

当你需要限制值只能从几个固定的选项中选取时：

from typing import Literal

# 只能传这几个字符串
def set_mode(mode: Literal["dark", "light", "auto"]) -> None:
    print(f"Mode set to {mode}")

set_mode("dark")   # ✅
set_mode("blue")   # ❌ IDE 报错

# 常用于 API 参数的状态过滤
OrderStatus = Literal["pending", "paid", "shipped", "delivered", "cancelled"]

def get_orders(status: OrderStatus) -> list[dict]:
    ...

6. Annotated —— FastAPI 的推荐方式

Annotated 是 Python 3.9+ 引入的，允许你在类型标注上附加元数据。FastAPI 推荐用这种方式：

from typing import Annotated

# 基础用法：Annotated[类型, 元数据]
from fastapi import FastAPI, Query, Path
from pydantic import Field

app = FastAPI()

# FastAPI 中：用 Annotated 给参数附加校验规则
# ge->大于等于 le->小于等于
@app.get("/items")
async def list_items(
    page: Annotated[int, Query(ge=1, description="页码")] = 1,
    size: Annotated[int, Query(ge=1, le=100, description="每页数量")] = 20,
):
    return {"page": page, "size": size}

@app.get("/items/{item_id}")
async def get_item(
    item_id: Annotated[int, Path(ge=1, description="商品 ID")],
):
    return {"item_id": item_id}

为什么 FastAPI 推荐 Annotated？ 因为在同一个参数上使用 Query() 和 Body() 等函数时，代码更干净，不会出现类型和默认值混在一起的混乱情况。

7. 实战练习

在 code/阶段0-前置基础/ 下创建 type_hints_practice.py，完成以下练习：

"""类型标注练习 —— 完成以下函数，确保类型标注正确"""

from typing import Optional, List, Dict

# 练习 1：为函数添加类型标注
def calculate_total(prices, quantities):
    """计算总价，prices 是单价列表，quantities 是数量列表"""
    total = 0
    for price, qty in zip(prices, quantities):
        total += price * qty
    return total


# 练习 2：dict 类型标注
def get_user_info(user_id):
    """根据用户 ID 返回用户信息字典"""
    users = {
        1: {"name": "Alice", "age": 25, "email": "alice@example.com"},
        2: {"name": "Bob", "age": 30, "email": "bob@example.com"},
    }
    return users.get(user_id)


# 练习 3：Optional 的使用
def parse_int(s):
    """将字符串转为 int，如果无法转换返回 None"""
    try:
        return int(s)
    except ValueError:
        return None


# 练习 4：Literal 类型
from typing import Literal

def create_user(name: str, role):
    """创建用户，role 只能是 'admin'、'editor'、'viewer'"""
    return {"name": name, "role": role}


# --- 写完以上代码后，用 mypy 检查 ---
# pip install mypy
# mypy type_hints_practice.py

8. 补充学习：dataclass 与 Pydantic 的对比

在进入 Pydantic 之前，先了解标准库的 dataclass：

from dataclasses import dataclass

@dataclass
class User:
    name: str
    age: int
    email: str = ""   # 有默认值的字段

# 自动生成 __init__, __repr__, __eq__ 等方法
user = User(name="Alice", age=25)
print(user)  # User(name='Alice', age=25, email='')

dataclass 缺点：不做数据验证。name=123 也能创建成功。Pydantic 就是来解决这个问题的。

9. 本章检查清单

完成以下所有项，再进入下一章：

  • 理解类型标注只影响静态检查，不影响运行时
  • 能正确标注：基本类型、List、Dict、Optional、Union
  • 知道 Python 3.10+ 的新语法（list[int], X | None）
  • 理解 Literal 的用途
  • 理解 Annotated 的作用 —— 给类型附加元数据
  • 完成实战练习中的所有函数标注
  • 安装 mypy 并运行，修复所有类型警告

10. 进入下一章前的准备

在终端中确认环境：

python --version          # 确认 >= 3.8（推荐 3.10+）
pip install mypy          # 静态类型检查工具
pip install pydantic      # 下一章需要

🎯 掌握类型标注后，你就理解了 FastAPI 自动解析参数和生成文档的底层魔法。0.2-异步编程基础 见！

0.2 异步编程基础

FastAPI 是一个异步 Web 框架。理解 async/await 是发挥其性能优势的前提。

1. 什么是并发？什么是并行？

概念 含义 类比 Python 实现
并发 (Concurrency) 交替执行多个任务 一个人同时下两盘棋，轮流走子 asyncio
并行 (Parallelism) 同时执行多个任务 两个人各自下一盘棋 multiprocessing 
# 并发：一个线程交替做两件事
# 并行：两个 CPU 核心同时做两件事

# Python 的 GIL 限制了多线程无法实现真正的并行
# asyncio 通过事件循环实现高效的并发（尤其是 IO 密集型任务）

2. 同步 vs 异步 — 直观感受

import time
import asyncio

# ===== 同步版本：一个接一个 =====
def fetch_data_sync(task_id: int) -> str:
    print(f"  [任务 {task_id}] 开始请求...")
    time.sleep(2)  # 模拟网络 IO
    print(f"  [任务 {task_id}] 完成！")
    return f"数据-{task_id}"

def main_sync():
    print("===== 同步模式 =====")
    start = time.time()
    for i in range(3):
        fetch_data_sync(i)
    print(f"总耗时: {time.time() - start:.2f}s\n")

main_sync()
# 输出：总耗时约 6 秒（3 个任务 × 2 秒）

# ===== 异步版本：同时发出请求 =====
async def fetch_data_async(task_id: int) -> str:
    print(f"  [任务 {task_id}] 开始请求...")
    await asyncio.sleep(2)  # 模拟异步 IO，不阻塞！
    print(f"  [任务 {task_id}] 完成！")
    return f"数据-{task_id}"

async def main_async():
    print("===== 异步模式 =====")
    start = time.time()
    # 同时启动 3 个任务
    results = await asyncio.gather(
        fetch_data_async(0),
        fetch_data_async(1),
        fetch_data_async(2),
    )
    print(f"总耗时: {time.time() - start:.2f}s")
    print(f"结果: {results}")

asyncio.run(main_async())
# 输出：总耗时约 2 秒（3 个任务并发执行！）

3. 核心概念

3.1 协程（Coroutine）

# async def 定义的函数返回一个协程对象
async def hello():
    return "Hello"

# 调用它不会执行，只是创建了一个协程对象
coro = hello()
print(type(coro))  # <class 'coroutine'>

# 必须用 await 或 asyncio.run() 来执行
result = asyncio.run(coro)
print(result)  # "Hello"

3.2 await — 交出控制权

async def demo():
    print("A")
    await asyncio.sleep(1)  # 交出控制权，让其他协程运行
    print("B")
    await asyncio.sleep(1)  # 再次交出控制权
    print("C")

# await 意味着："我暂时不做事了，其他人可以先做"
# 只能在 async def 函数内部使用 await

3.3 事件循环（Event Loop）

# asyncio.run() 做了三件事：
# 1. 创建一个事件循环
# 2. 把你的协程放进循环
# 3. 运行循环直到协程完成

async def main():
    print("在事件循环中运行")

asyncio.run(main())  # 这是标准入口

4. 常用 asyncio API

import asyncio

async def task(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} 完成"

async def main():
    # === gather: 并发执行，等所有完成返回结果列表 ===
    results = await asyncio.gather(
        task("A", 2),
        task("B", 1),
        task("C", 3),
    )
    print(f"gather 结果: {results}")
    # ['A 完成', 'B 完成', 'C 完成'] — B 先完成但顺序保持

    # === create_task: 立即调度，不阻塞 ===
    t = asyncio.create_task(task("后台任务", 5))
    print("主协程继续干活，不等后台任务...")
    # 做点别的事...
    result = await t  # 需要结果时再等
    print(result)

    # === wait_for: 超时控制 ===
    try:
        result = await asyncio.wait_for(task("慢任务", 10), timeout=3)
    except asyncio.TimeoutError:
        print("任务超时了！")

    # === as_completed: 哪个先完成先处理哪个 ===
    tasks = [task(f"X-{i}", delay) for i, delay in enumerate([3, 1, 2])]
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"最先完成的: {result}")

asyncio.run(main())

5. 异步上下文管理器

# 同步写法
with open("file.txt") as f:
    data = f.read()

# 异步写法（如数据库连接、HTTP 会话）
import asyncio

class AsyncDB:
    async def __aenter__(self):
        print("连接数据库...")
        await asyncio.sleep(0.5)
        return self

    async def __aexit__(self, *args):
        print("关闭数据库连接...")
        await asyncio.sleep(0.1)

    async def query(self, sql: str):
        await asyncio.sleep(0.3)
        return f"执行: {sql}"

async def main():
    async with AsyncDB() as db:
        result = await db.query("SELECT * FROM users")
        print(result)
    # 退出 with 时自动关闭连接

asyncio.run(main())

6. 一个真实场景：并发 HTTP 请求

# 先安装: pip install httpx
import asyncio
import time
import httpx

URLS = [
    "https://httpbin.org/delay/2",
    "https://httpbin.org/delay/2",
    "https://httpbin.org/delay/2",
]

# ===== 同步版本 =====
def fetch_sync(url: str):
    with httpx.Client() as client:
        return client.get(url)

def sync_demo():
    start = time.time()
    for url in URLS:
        resp = fetch_sync(url)
        print(f"状态: {resp.status_code}")
    print(f"同步耗时: {time.time() - start:.2f}s")

# ===== 异步版本 =====
async def fetch_async(url: str):
    async with httpx.AsyncClient() as client:
        return await client.get(url)

async def async_demo():
    start = time.time()
    tasks = [fetch_async(url) for url in URLS]
    results = await asyncio.gather(*tasks)
    for resp in results:
        print(f"状态: {resp.status_code}")
    print(f"异步耗时: {time.time() - start:.2f}s")

# 运行对比
sync_demo()                # ~6s
asyncio.run(async_demo())  # ~2s

7. 最常见陷阱 ⚠️

陷阱 1：在协程中调用同步阻塞函数

import time

async def bad_example():
    """❌ 不要这样做！"""
    time.sleep(5)  # 同步阻塞！会卡住整个事件循环！
    # 应该用 await asyncio.sleep(5)

async def good_example():
    """✅ 用异步替代品"""
    await asyncio.sleep(5)

# 如果必须调用同步函数，用线程池：
async def run_sync_in_thread():
    loop = asyncio.get_running_loop()
    # 在线程池中运行同步函数，不阻塞事件循环
    result = await loop.run_in_executor(None, time.sleep, 2)

陷阱 2：忘记 await

async def get_data():
    await asyncio.sleep(1)
    return "data"

async def bug():
    result = get_data()  # ❌ 忘记 await！result 是一个协程对象，不是字符串
    # RuntimeWarning: coroutine 'get_data' was never awaited
    print(result)  # <coroutine object get_data at 0x...>

async def fixed():
    result = await get_data()  # ✅
    print(result)  # "data"

陷阱 3：async def 但没有 await

async def no_await():
    # ⚠️ 这个函数虽然是 async，但内部没有 await
    # 它仍然会同步执行，不能获得异步的好处
    return 1 + 1

8. FastAPI 中的异步

FastAPI 中你看到最多的模式：

from fastapi import FastAPI

app = FastAPI()

@app.get("/sync-endpoint")
def sync_endpoint():                    # 同步路径操作函数
    return {"message": "hello"}         # FastAPI 在线程池中运行它

@app.get("/async-endpoint")
async def async_endpoint():             # 异步路径操作函数
    await asyncio.sleep(0.1)            # 异步 IO 不会阻塞
    return {"message": "hello"}

# 规则：
# 1. 如果你有异步 IO（数据库、HTTP 请求），用 async def + await
# 2. 如果全是 CPU 计算（没有 IO），用 def（同步）就够了
# 3. 不要在 async def 中调用同步阻塞函数！

9. 实战练习

创建 code/阶段0-前置基础/async_practice.py：

"""异步编程练习"""
import asyncio
import time

# 练习 1：模拟并发爬虫
# 写一个异步函数，并发"抓取" 5 个 URL（用 asyncio.sleep 模拟网络延迟）
# 每个 URL 的延迟随机在 1-3 秒之间
# 用 asyncio.gather 并发执行，最后打印总耗时

# 你的代码：


# 练习 2：超时控制
# 有一个函数可能执行很长时间，你需要给它的执行加上 3 秒超时
# 如果超时，打印提示信息并返回默认值

# 你的代码：


# 练习 3：生产者-消费者模式
# 一个生产者协程每秒产生一个数字（1-5）
# 一个消费者协程消费这些数字（每次消费需要 0.5 秒）
# 使用 asyncio.Queue 实现

# 你的代码：

10. 本章检查清单

  • 理解并发 vs 并行的区别
  • 能解释 async def、await、事件循环的关系
  • 能用 asyncio.gather() 并发执行多个协程
  • 理解 asyncio.create_task() 和 asyncio.wait_for()
  • 能在 httpx 中对比同步和异步的性能差异
  • 知道三个常见陷阱：同步阻塞、忘记 await、伪异步
  • 理解 FastAPI 中何时用 def、何时用 async def
  • 完成实战练习的 3 个题目

🎯 异步是 FastAPI 高性能的基石。接下来学 HTTP 协议基础，然后你就会明白 FastAPI 的 API 设计为什么是现在的样子。

0.3 HTTP 协议基础

FastAPI 是一个 HTTP API 框架。深入理解 HTTP 协议，你才能写出符合标准的、优雅的 API。

1. HTTP 是什么？

HTTP（HyperText Transfer Protocol）是 Web 的通用语言。浏览器和服务器之间、前端和后端之间、微服务之间，都通过 HTTP 通信。

客户端（浏览器/Postman/前端）           服务器（FastAPI）
         │                                    │
         │──── GET /users HTTP/1.1 ──────────>│
         │                                    │ 处理请求
         │<──── HTTP/1.1 200 OK ──────────────│
         │      Content-Type: application/json│
         │      {"users": [...]}              │

2. HTTP 方法（Verbs）

方法 含义 幂等性 典型用途
GET 读取资源 ✅ 幂等 查询用户列表、获取文章详情
POST 创建资源 ❌ 不幂等 注册用户、创建订单
PUT 完整替换资源 ✅ 幂等 更新用户全部信息
PATCH 部分更新资源 ❌ 不幂等 只修改用户邮箱
DELETE 删除资源 ✅ 幂等 删除文章、删除评论
HEAD 只获取响应头 ✅ 幂等 检查资源是否存在
OPTIONS 查询支持的方法 ✅ 幂等 CORS 预检请求

幂等性：同一个操作执行一次和执行多次，结果一样。GET 查询 10 次结果相同；POST 创建 10 次得到 10 条不同的记录。

3. 状态码（Status Codes）

2xx — 成功

200 OK           # 请求成功（GET/PUT/PATCH）
201 Created      # 创建成功（POST）— 返回新资源的 URL
204 No Content   # 成功但没有返回内容（DELETE）

3xx — 重定向

301 Moved Permanently  # 资源永久迁移
302 Found              # 临时重定向
304 Not Modified       # 资源未修改（配合缓存使用）

4xx — 客户端错误

400 Bad Request        # 请求格式有误（参数校验失败）
401 Unauthorized       # 需要登录（未认证）
403 Forbidden          # 没有权限（已认证但无权限）
404 Not Found          # 资源不存在
405 Method Not Allowed # HTTP 方法不对
409 Conflict           # 资源冲突（如重复注册）
422 Unprocessable Entity # 请求格式正确但语义有问题（FastAPI 常用）
429 Too Many Requests  # 请求太频繁（触发限流）

5xx — 服务端错误

500 Internal Server Error  # 服务器内部错误（代码 bug）
502 Bad Gateway            # 上游服务返回无效响应
503 Service Unavailable    # 服务暂时不可用（维护中）
504 Gateway Timeout        # 上游服务超时

4. URL 结构

https://api.example.com:443/v1/users/42?include=posts&page=1
\____/  \______________/ \_/ \_____________________________/
协议       主机名（域名）   端口          路径 + 查询参数

协议      → https
主机      → api.example.com
端口      → 443（https 默认，通常省略）
路径      → /v1/users/42        （/v1 是版本，/users 是资源，/42 是标识）
查询参数  → include=posts&page=1

路径参数 vs 查询参数

# 路径参数：资源唯一标识（必选）
GET /users/42          # 获取 ID=42 的用户
GET /articles/123      # 获取 ID=123 的文章
GET /users/42/posts    # 获取用户 42 的所有文章

# 查询参数：过滤、排序、分页（可选，有默认值）
GET /users?status=active       # 筛选状态为 active 的用户
GET /users?page=1&size=20      # 分页
GET /users?sort=created_at     # 排序
GET /users?q=alice             # 搜索

5. 请求与响应的结构

完整的 HTTP 请求

POST /api/v1/users HTTP/1.1                    ← 请求行
Host: api.example.com                          ← 请求头
Content-Type: application/json                 ← 请求头
Authorization: Bearer eyJhbGciOi...            ← 请求头
Accept: application/json                       ← 请求头

{                                               ← 请求体（空行之后）
  "name": "Alice",
  "email": "alice@example.com",
  "age": 25
}

完整的 HTTP 响应

HTTP/1.1 201 Created                           ← 状态行
Content-Type: application/json                 ← 响应头
Content-Length: 45                             ← 响应头
Location: /api/v1/users/123                    ← 响应头（新建资源的 URL）

{                                               ← 响应体
  "id": 123,
  "name": "Alice",
  "email": "alice@example.com"
}

6. 常用请求头

请求头 作用 示例
Content-Type 请求体的数据格式 application/json
Authorization 认证凭据 Bearer <token>
Accept 客户端期望的响应格式 application/json
User-Agent 客户端标识 Mozilla/5.0 ...
Cookie 发送 Cookie session_id=abc123
Cache-Control 缓存策略 no-cache

常用响应头

响应头 作用 示例
Content-Type 响应体的数据格式 application/json; charset=utf-8
Set-Cookie 设置 Cookie session=abc; HttpOnly
Cache-Control 缓存策略 max-age=3600
Location 重定向目标 /api/v1/users/123
Access-Control-Allow-Origin CORS * 或特定域名

7. RESTful API 设计原则

REST（Representational State Transfer）是一套 API 设计约定，不是标准，但业界广泛遵循：

# ✅ 好的 RESTful 设计
GET    /api/v1/users           # 获取用户列表
GET    /api/v1/users/42        # 获取用户 42
POST   /api/v1/users           # 创建新用户
PUT    /api/v1/users/42        # 完整更新用户 42
PATCH  /api/v1/users/42        # 部分更新用户 42
DELETE /api/v1/users/42        # 删除用户 42

# 嵌套资源
GET    /api/v1/users/42/posts          # 用户 42 的所有文章
GET    /api/v1/users/42/posts/5        # 用户 42 的第 5 篇文章
GET    /api/v1/users/42/posts/5/comments  # 该文章的所有评论

# ❌ 不好的设计
GET  /api/v1/getAllUsers            # 动词不要出现在 URL 中
POST /api/v1/users/create           # 避免动词
GET  /api/v1/users?action=delete    # 不要用查询参数区分操作
DELETE /api/v1/users                # 危险！不要批量删除

核心原则

  1. 用名词，不用动词：资源是名词（users, orders, articles）
  2. 用 HTTP 方法表达操作：GET=获取，POST=创建，PUT/PATCH=更新，DELETE=删除
  3. 层级关系用 URL 嵌套：/users/42/orders/5
  4. 用查询参数做过滤、排序、分页
  5. 版本化你的 API：/v1/, /v2/
  6. 用复数名词：/users 而不是 /user

8. JSON 数据格式

API 通信的事实标准：

{
  "id": 1,
  "name": "Alice",
  "email": "alice@example.com",
  "is_active": true,
  "roles": ["user", "editor"],
  "profile": {
    "bio": "Python developer",
    "website": "https://alice.dev"
  },
  "created_at": "2024-01-15T08:30:00Z"
}

FastAPI 会自动将 Python 字典和 Pydantic 模型序列化为 JSON。

9. 用浏览器 DevTools 分析真实 API

动手环节：打开浏览器 F12 → Network 标签，访问任意网站，观察 API 请求。

以 GitHub API 为例

# 用 curl 测试（或浏览器直接打开）
curl https://api.github.com/users/tiangolo

# 观察响应：
# - 状态码 200
# - Content-Type: application/json
# - 响应体中包含 login, id, avatar_url 等字段

试着分析：

  • 请求方法是什么？
  • URL 的路径参数和查询参数分别是什么？
  • 响应状态码是多少？
  • 响应头中有什么信息？

10. FastAPI 中的映射

你在 FastAPI 中写的每个路由，本质都是在定义 HTTP 处理规则：

from fastapi import FastAPI, Path, Query, Header

app = FastAPI()

# 这对应 HTTP GET /users/{user_id}
@app.get("/users/{user_id}", status_code=200)          # ← 方法  + 路径 + 默认状态码
async def get_user(
    user_id: int = Path(..., ge=1),                    # ← 路径参数，来自 URL 路径
    include_posts: bool = Query(False),                # ← 查询参数，来自 ?include_posts=true
    user_agent: str = Header(default=""),              # ← 请求头，来自 User-Agent
):
    return {"user_id": user_id, "include_posts": include_posts}
HTTP 概念 FastAPI 写法
URL 路径 @app.get("/path") 第一个参数
路径参数 Path(...)
查询参数 Query(...) 或直接写函数参数
请求头 Header(...)
请求体 Pydantic 模型作为函数参数
状态码 status_code 参数
响应体 return 的值

11. 实战练习

创建 code/阶段0-前置基础/http_practice.md，完成以下题目：

练习 1：设计一个电商 API 的 URL 结构
  为以下功能设计 RESTful URL（包括方法 + 路径）：
  - 获取商品列表（支持分页和分类筛选）
  - 获取单个商品详情
  - 创建订单
  - 查看用户的所有订单
  - 取消订单
  - 给商品添加评价

练习 2：状态码场景判断
  以下情况应该返回什么状态码？
  - 用户登录时密码错误
  - 请求的商品 ID 不存在
  - 用户没有绑定手机号却尝试发帖
  - 服务器数据库挂了
  - 短时间内提交了 100 次同样的请求

练习 3：用 curl 或浏览器测试 3 个公开 API
  - https://api.github.com/users/你的用户名
  - https://httpbin.org/get?name=test
  - https://jsonplaceholder.typicode.com/posts/1
  记录每个 API 的：方法、状态码、Content-Type、响应结构

12. 本章检查清单

  • 能说出 6 种 HTTP 方法及各自用途
  • 理解幂等性，能区分 POST 和 PUT
  • 能根据场景选择合适的 HTTP 状态码
  • 能区分路径参数和查询参数
  • 理解 RESTful URL 设计原则
  • 能读懂 HTTP 请求和响应的结构
  • 知道 JSON 是 API 数据交换的标准格式
  • 用手测试过至少 3 个公开 API
  • 知道 FastAPI 的路由如何映射到 HTTP 概念

🎯 HTTP 是 API 的通用语言。下一章学 Pydantic，它是 FastAPI 数据验证的核心引擎。

0.4 Pydantic 基础

Pydantic 是 FastAPI 的"发动机"——数据校验、序列化、文档生成全靠它。V2 版本（2023 年发布）性能提升 5-50 倍。

1. Pydantic 解决什么问题？

# ❌ 传统方式：手动校验，又臭又长
def create_user(data: dict):
    if not isinstance(data.get("name"), str):
        raise ValueError("name 必须是字符串")
    if len(data["name"]) < 2:
        raise ValueError("name 至少 2 个字符")
    if not isinstance(data.get("age"), int):
        raise ValueError("age 必须是整数")
    if data["age"] < 0 or data["age"] > 150:
        raise ValueError("age 必须在 0-150 之间")
    # ... 还有 email 格式、手机号格式……写到崩溃
    return {"name": data["name"], "age": data["age"]}

# ✅ Pydantic：声明式校验，干净优雅
from pydantic import BaseModel, Field

class UserCreate(BaseModel):
    name: str = Field(min_length=2, max_length=50, description="用户名")
    age: int = Field(ge=0, le=150, description="年龄")

# 一行校验，失败时自动生成清晰的错误信息
user = UserCreate(name="A", age=200)
# ValidationError:
#   name: String should have at least 2 characters
#   age: Input should be less than or equal to 150

2. 第一个 Pydantic 模型

from pydantic import BaseModel
from datetime import datetime

class User(BaseModel):
    id: int
    name: str
    email: str
    age: int = 18                      # 有默认值的字段
    is_active: bool = True
    created_at: datetime = None        # 可以设默认值为 None

# 创建实例 —— 自动校验
user = User(
    id=1,
    name="Alice",
    email="alice@example.com",
    age=25,
)
print(user.name)          # "Alice"
print(user.age)           # 25
print(user.is_active)     # True（使用默认值）

# 字典 ↔ 模型 互转
print(user.model_dump())
# {'id': 1, 'name': 'Alice', 'email': 'alice@example.com', 'age': 25, ...}

# JSON 序列化
print(user.model_dump_json())
# '{"id":1,"name":"Alice","email":"alice@example.com","age":25,...}'

重要：V2 API 对应关系

V1（旧版） V2（新版，推荐） 用途
.dict() .model_dump() 转为字典
.json() .model_dump_json() 转为 JSON 字符串
.parse_obj(d) .model_validate(d) 从字典创建模型
.parse_raw(s) .model_validate_json(s) 从 JSON 字符串创建
.schema() .model_json_schema() 生成 JSON Schema

3. Field —— 字段校验

from pydantic import BaseModel, Field
from typing import Optional

class Product(BaseModel):
    name: str = Field(
        min_length=1,
        max_length=100,
        description="商品名称",
    )
    price: float = Field(
        gt=0,                              # greater than 0
        le=999999.99,                      # less than or equal to
        description="价格",
    )
    stock: int = Field(
        ge=0,                              # greater than or equal to
        default=0,
        description="库存数量",
    )
    tags: list[str] = Field(
        default_factory=list,             # 可变默认值要用 default_factory！
        max_length=10,                     # 最多 10 个标签
        description="商品标签",
    )
    description: Optional[str] = Field(
        default=None,
        max_length=1000,
    )

# 测试校验
try:
    Product(name="", price=-10, stock=-5)
except Exception as e:
    print(e)
    # 3 个校验错误同时报告！

Field 常用参数速查

参数 适用类型 含义
gt / ge int, float 大于 / 大于等于
lt / le int, float 小于 / 小于等于
min_length / max_length str, list 最小 / 最大长度
pattern str 正则匹配
default 任意 默认值
default_factory 任意 生成默认值的函数（用于可变类型）
description 任意 文档说明（Swagger 中显示）
examples 任意 示例值
alias 任意 字段别名（JSON 字段名和 Python 属性名不同时）

4. 嵌套模型

from pydantic import BaseModel
from typing import List

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

class User(BaseModel):
    name: str
    email: str
    address: Address                    # 嵌套一个模型
    tags: List[str] = []

# 创建时自动递归校验
user = User(
    name="Alice",
    email="alice@example.com",
    address={
        "street": "123 Main St",        # 字典会自动转为 Address
        "city": "Beijing",
        "zip_code": "100000",
    },
    tags=["python", "fastapi"],
)

print(user.address.city)            # "Beijing"
print(user.model_dump())
# {
#   "name": "Alice",
#   "email": "alice@example.com",
#   "address": {"street": "123 Main St", "city": "Beijing", "zip_code": "100000"},
#   "tags": ["python", "fastapi"]
# }

5. 模型继承

class UserBase(BaseModel):
    """基础字段：创建和读取都需要的"""
    name: str
    email: str

class UserCreate(UserBase):
    """创建用户：基础字段 + 密码"""
    password: str = Field(min_length=8)

class UserResponse(UserBase):
    """返回用户：基础字段 + 数据库 ID + 时间戳"""
    id: int
    created_at: str     # 实际项目中用 datetime

class UserInDB(UserResponse):
    """数据库存储：返回字段 + 哈希密码"""
    hashed_password: str

# 清晰的继承关系：
# UserBase → UserCreate (输入)
# UserBase → UserResponse (输出) → UserInDB (持久化)

6. 自定义验证器

field_validator（V2 新语法，替代 V1 的 validator）

from pydantic import BaseModel, field_validator, Field
import re

class UserRegister(BaseModel):
    username: str = Field(min_length=3, max_length=20)
    email: str
    password: str = Field(min_length=8)
    phone: str

    @field_validator("username")
    @classmethod
    def username_must_be_alphanumeric(cls, v: str) -> str:
        """用户名只能包含字母、数字和下划线"""
        if not re.match(r"^[a-zA-Z0-9_]+$", v):
            raise ValueError("用户名只能包含字母、数字和下划线")
        return v

    @field_validator("email")
    @classmethod
    def email_must_be_valid(cls, v: str) -> str:
        """简单邮箱校验"""
        if "@" not in v or "." not in v.split("@")[-1]:
            raise ValueError("邮箱格式不正确")
        return v.lower()  # 统一转小写

    @field_validator("phone")
    @classmethod
    def phone_must_be_chinese(cls, v: str) -> str:
        """校验中国大陆手机号"""
        if not re.match(r"^1[3-9]\d{9}$", v):
            raise ValueError("请输入有效的手机号")
        return v

model_validator（跨字段校验）

from pydantic import BaseModel, model_validator
from typing import Self

class PasswordChange(BaseModel):
    password: str
    password_confirm: str

    @model_validator(mode="after")
    def passwords_must_match(self) -> Self:
        """密码和确认密码必须一致"""
        if self.password != self.password_confirm:
            raise ValueError("两次输入的密码不一致")
        return self

7. JSON Schema 自动生成

Pydantic 模型可以自动生成 JSON Schema，这就是 FastAPI OpenAPI 文档的底层机制：

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(description="商品名称")
    price: float = Field(gt=0, description="价格")
    is_available: bool = True

# 查看生成的 JSON Schema
print(Item.model_json_schema())
# {
#   "title": "Item",
#   "type": "object",
#   "properties": {
#     "name": {"title": "Name", "type": "string", "description": "商品名称"},
#     "price": {"title": "Price", "type": "number", "description": "价格", "exclusiveMinimum": 0},
#     "is_available": {"title": "Is Available", "type": "boolean", "default": true}
#   },
#   "required": ["name", "price"]
# }

8. Pydantic 在 FastAPI 中的角色

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

# Pydantic 模型定义请求体
class UserCreate(BaseModel):
    name: str = Field(min_length=2)
    email: str
    age: int = Field(ge=0, le=150)

# Pydantic 模型定义响应体
class UserResponse(BaseModel):
    id: int
    name: str
    email: str

@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate):      # ← 自动校验请求体
    # user 是已经校验过的 UserCreate 实例
    return {"id": 1, **user.model_dump()}     # ← 自动按 UserResponse 过滤输出

# 访问 /docs 你会看到自动生成的：
# - UserCreate 作为请求体 Schema
# - UserResponse 作为响应体 Schema
# - 所有 Field 的 description、min_length 等都在文档中展示

9. 实战练习

创建 code/阶段0-前置基础/pydantic_practice.py：

"""Pydantic 练习"""
from pydantic import BaseModel, Field, field_validator, model_validator
from typing import List, Optional
from enum import Enum

# 练习 1：商品模型
# 定义 Product 模型：
# - name: str（必填，1-100 字符）
# - price: float（必填，> 0）
# - category: str（必填，只能是 "electronics", "clothing", "food" 之一）
# - tags: list[str]（可选，默认空列表，最多 5 个标签）
# - stock: int（可选，默认 0，必须 >= 0）

# 你的代码：


# 练习 2：嵌套模型 + 自定义校验
# 定义 OrderItem（商品 ID + 数量，数量 >= 1）
# 定义 Order（订单号 + 商品列表 + 收货地址）
# 用 model_validator 保证：订单至少包含 1 个商品

class Address(BaseModel):
    province: str
    city: str
    detail: str

# 你的 OrderItem 和 Order：


# 练习 3：模拟 FastAPI 的请求/响应分离
# UserCreate 模型（输入）：name, email, password
# UserResponse 模型（输出）：id, name, email（不暴露密码！）
# 写一个函数，接收 dict 创建 UserCreate，然后模拟返回 UserResponse

# 你的代码：

10. 关键概念速记

# Pydantic 核心流程
dict/JSON  ──> .model_validate() ──> Pydantic 实例（已校验） ──> .model_dump() ──> dict
                                  │
                      发生错误 → ValidationError（包含所有字段的错误详情）

# 三部曲
# 1. 定义模型（声明结构 + 校验规则）
class Model(BaseModel):
    field: type = Field(...)

# 2. 创建实例（自动校验）
obj = Model(**raw_data)

# 3. 使用或导出
obj.field              # 属性访问
obj.model_dump()       # → dict
obj.model_dump_json()  # → JSON 字符串

11. 本章检查清单

  • 理解 Pydantic 相对于手动校验的优势
  • 能定义模型：字段类型 + Field 校验参数
  • 知道 V2 API（model_dump, model_validate）
  • 能创建嵌套模型
  • 能用模型继承分离输入/输出/存储
  • 能写 field_validator 做单字段校验
  • 能写 model_validator 做跨字段校验
  • 理解 Pydantic 自动生成 JSON Schema 的原理
  • 知道 Pydantic 模型在 FastAPI 中的请求/响应角色
  • 完成实战练习的 3 个题目

🎯 阶段 0 完成！ 你已经掌握了类型标注、异步编程、HTTP 协议和 Pydantic。现在可以自信地进入 阶段 1：FastAPI 核心 了。打开 0.1-Python类型标注.md 复习检查清单，确认全部勾选后，进入第一章！

# fastapi # 学习 # python
Logo
AtomGit开源社区

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

更多推荐

cover

AtomGit 首发开源｜光轮智能发布全球首个十万小时级全模态人类行为开源数据集 EgoSuite-Open100K

avatar AtomGit开源社区
cover

构筑赋能金融行业应用 Agent｜20 强决战北京，一场金融 Agent 的硬核答卷

avatar AtomGit开源社区
cover

AtomCode 全新上线万亿代码模型 LongCat-2.0。限时 7 天免费体验！

avatar AtomGit开源社区