Zhimalab

DAY 1

初识 FastAPI

从零开始，用 30 行代码搭起你的第一个 API 服务，体验自动文档的魅力。

🎯 目标：跑通第一个 API⏱️ 预计：45 分钟📦 依赖：Python 3.8+

1FastAPI 是什么？

FastAPI 是一个现代、高性能的 Python Web 框架，由 Sebastián Ramírez 于 2018 年创建。它基于 Starlette（异步 ASGI 框架）和 Pydantic（数据验证库），天生支持异步编程。

⚡

极致性能

与 Node.js、Go 并驾齐驱，是 Python 中最快的框架之一。

📖

自动文档

自动生成 Swagger UI 和 ReDoc 交互文档，开箱即用。

🔐

类型安全

基于 Python 类型注解，请求/响应自动验证与转换。

🚀

开发高效

代码量少、提示友好，编辑器自动补全体验极佳。

FastAPI vs Flask vs Django？ FastAPI 适合构建 API 服务和微服务；Flask 轻量灵活但同步为主；Django 全栈但偏重。如果你要做纯 API，FastAPI 是当下的最佳选择。

2安装与第一个程序

首先创建虚拟环境并安装 FastAPI 和 ASGI 服务器 Uvicorn：

终端

# 创建项目目录
mkdir myapi && cd myapi

# 创建虚拟环境
python -m venv venv
source venv/bin/activate    # Windows: venv\Scripts\activate

# 安装 FastAPI 和 Uvicorn
pip install fastapi uvicorn[standard]

创建主程序文件 main.py：

main.py

from fastapi import FastAPI

app = FastAPI(title="我的第一个 API", version="1.0.0")

@app.get("/")
def read_root():
    return {"message": "Hello, FastAPI!", "status": "running"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "q": q}

启动开发服务器：

终端

uvicorn main:app --reload

# main:app 含义：
#   main -> 文件名 main.py
#   app  -> 实例变量名 app
# --reload 开启热重载，改代码自动重启

启动后你会看到类似输出：

输出

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Application startup complete.

3自动文档——杀手级特性

FastAPI 自动生成两套交互式 API 文档，无需写一行额外代码：

文档地址	名称	特点
`http://localhost:8000/docs`	Swagger UI	可直接在浏览器里测试每个接口
`http://localhost:8000/redoc`	ReDoc	布局优雅，适合做 API 参考文档
`http://localhost:8000/openapi.json`	OpenAPI Schema	机器可读的 JSON 规范

动手试试：打开 http://localhost:8000/docs，找到 GET /items/{item_id}，点击 "Try it out"，输入 item_id=42 和 q=hello，点击 Execute——你会立刻看到 JSON 响应！

4代码逐行解析

main.py 关键行

app = FastAPI(title="...")     # 创建应用实例，可设标题、描述、版本
@app.get("/")                 # 路由装饰器：GET 请求，路径 "/"
def read_root():               # 同步处理函数（也支持 async def）
    return {"message": "..."}  # 返回 dict，FastAPI 自动转 JSON
@app.get("/items/{item_id}")  # 路径参数用 {} 包裹
def read_item(item_id: int):   # 类型注解 int → 自动校验和转换
    ...                        # 非 int 会自动返回 422 错误

同步 vs 异步：用 def 声明的函数运行在线程池中；用 async def 声明的函数运行在事件循环中。如果函数内有 await 调用（如异步数据库、HTTP 请求），必须用 async def。

随堂测验

理解检查

在 uvicorn main:app --reload 命令中，main:app 的含义是什么？

A main 模块的 app 配置文件

B main.py 文件中的 app 变量（FastAPI 实例）

C 名为 main 的数据库和名为 app 的表

D main 应用的 app 路由前缀

✏️ 练习

添加一个 /hello 接口

在 main.py 中添加一个 GET /hello 接口，返回 {"greeting": "你好，世界"}。写完后点击查看参考答案。

main.py

@app.get("/hello")
def hello():
    return {"greeting": "你好，世界"}

保存后 --reload 会自动重启，访问 http://localhost:8000/hello 即可看到结果。

🎯

第一天完成！

你已经搭起了 FastAPI 开发环境，理解了路由、自动文档和基本结构。

DAY 2

路由与参数

掌握路径参数、查询参数和自动验证——FastAPI 类型系统的核心魅力。

🎯 目标：精通参数传递⏱️ 预计：50 分钟

1路径参数

路径参数用花括号 {} 声明，FastAPI 会根据类型注解自动校验和转换：

main.py

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id}

# 访问 /users/42    -> {"user_id": 42}（自动转 int）
# 访问 /users/abc   -> 422 错误，提示 "value is not a valid integer"

路径中可以有多个参数，顺序和声明顺序一致：

main.py

@app.get("/users/{user_id}/posts/{post_id}")
async def get_post(user_id: int, post_id: int):
    return {"user_id": user_id, "post_id": post_id}

路由顺序很重要！如果你同时有 /users/me 和 /users/{user_id}，必须把 /users/me 写在前面，否则 me 会被当作 user_id 传入，触发类型校验失败。

2查询参数

声明在路径之外的、有默认值或类型注解的参数，会被识别为查询参数：

main.py

# 模拟商品列表接口
items_db = ["苹果", "香蕉", "橙子", "葡萄", "西瓜"]

@app.get("/items")
async def list_items(skip: int = 0, limit: int = 10):
    return items_db[skip : skip + limit]

# /items               -> skip=0, limit=10（使用默认值）
# /items?skip=2        -> 跳过前2个
# /items?skip=1&limit=3 -> 跳过1个，取3个

必选 vs 可选查询参数：没有默认值的参数是必选的（不传会报错）；有默认值的是可选的。想要一个可选参数但无特定默认值，使用 Optional[str] = None。

3参数验证

使用 Query 可以对查询参数添加验证约束：

main.py

from fastapi import Query

@app.get("/search")
async def search(
    q: str = Query(
        min_length=3,        # 最少3个字符
        max_length=50,       # 最多50个字符
        pattern="^[\w ]+$",  # 正则：只允许字母数字下划线和空格
        description="搜索关键词"
    ),
    page: int = Query(default=1, ge=1),     # >= 1
    size: int = Query(default=10, ge=1, le=100)  # 1~100
):
    return {"q": q, "page": page, "size": size}

验证类型	关键字参数	适用类型
范围	`gt, ge, lt, le`	int, float
长度	`min_length, max_length`	str
正则	`pattern`	str
路径参数验证	用 `Path()` 替代 `Query()`	路径参数

🔌 互动模拟器：体验参数校验

模拟对 /search 接口的请求，尝试不同输入看返回结果

GET /search

q (关键词)

page

size

等待发送请求...

4路径参数 vs 查询参数

特征	路径参数	查询参数
声明方式	`/items/{id}`	函数参数有默认值
URL 位置	路径中	`?key=value`
是否必选	必选	有默认值则可选
语义	标识资源	过滤/排序/分页
验证器	`Path()`	`Query()`

随堂测验

理解检查

以下哪个 URL 能正确匹配 @app.get("/items/{item_id}") 且 item_id: int，同时传入查询参数 q？

A /items?q=test/42

B /items/42?q=test

C /items/42&q=test

D /items?item_id=42&q=test

✏️ 练习

实现分页查询接口

实现 GET /products，支持查询参数 category（可选，字符串）、min_price（可选，≥0 的浮点数）、page（默认1，≥1）、page_size（默认10，1~50）。返回这些参数的 JSON。

main.py

from fastapi import Query
from typing import Optional

@app.get("/products")
async def list_products(
    category: Optional[str] = None,
    min_price: float = Query(default=None, ge=0),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=10, ge=1, le=50),
):
    return {
        "filters": {"category": category, "min_price": min_price},
        "pagination": {"page": page, "page_size": page_size},
    }

🎯

第二天完成！

你已经掌握了路径参数、查询参数和参数验证机制。

DAY 3

请求体与 Pydantic 模型

用 Pydantic 定义数据模型，让请求体验证变得优雅而强大。

🎯 目标：掌握数据建模⏱️ 预计：55 分钟

1Pydantic 模型基础

Pydantic 是 FastAPI 的数据验证核心。你定义一个继承 BaseModel 的类，声明字段和类型，Pydantic 自动完成验证、转换和文档生成。

main.py

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(min_length=1, max_length=100, description="商品名称")
    price: float = Field(gt=0, description="价格，必须大于0")
    is_offer: bool = False
    tags: list[str] = []  # 默认空列表

# 这个模型自动出现在 /docs 的请求体 Schema 中

2接收请求体

把模型作为函数参数的类型注解，FastAPI 就知道要从请求体读取 JSON：

main.py

from fastapi import FastAPI

app = FastAPI()

@app.post("/items/")
async def create_item(item: Item):
    return {
        "received": item,
        "with_tax": item.price * 1.08,
    }

# 请求体（POST http://localhost:8000/items/）：
# {
#   "name": "键盘",
#   "price": 299.0,
#   "is_offer": true,
#   "tags": ["电子产品", "外设"]
# }

FastAPI 怎么知道参数是路径参数、查询参数还是请求体？规则很简单：①出现在路径 {} 中的 → 路径参数；②是 Pydantic 模型类型 → 请求体；③其他 → 查询参数。

3混合使用参数

路径参数、查询参数、请求体可以在同一个函数中同时使用：

main.py

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,              # 路径参数
    item: Item,                # 请求体
    q: str | None = None       # 查询参数
):
    result = {"item_id": item_id, **item.model_dump()}
    if q:
        result["q"] = q
    return result

4数据验证实战

Pydantic 会自动校验类型和约束，验证失败返回结构化的 422 错误：

验证失败示例

// 请求：POST /items/  body: {"name": "", "price": -5}
// 响应 422：
{
  "detail": [
    {
      "type": "string_too_short",
      "loc": ["body", "name"],
      "msg": "String should have at least 1 character",
      "input": ""
    },
    {
      "type": "greater_than",
      "loc": ["body", "price"],
      "msg": "Input should be greater than 0",
      "input": -5
    }
  ]
}

5嵌套模型

Pydantic 模型可以嵌套，表达复杂的数据结构：

main.py

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

class User(BaseModel):
    name: str
    age: int = Field(ge=0, le=150)
    address: Address        # 嵌套模型
    hobbies: list[str] = []

@app.post("/users")
async def create_user(user: User):
    return {"created": user}

# 请求体示例：
# {
#   "name": "张三",
#   "age": 20,
#   "address": {"city": "北京", "street": "中关村", "zip_code": "100080"},
#   "hobbies": ["编程", "音乐"]
# }

model_dump() vs dict()：Pydantic v2 用 model_dump() 将模型转为字典（v1 的 .dict() 已弃用）。model_json_schema() 可获取 JSON Schema。

随堂测验

理解检查

在 async def update_item(item_id: int, item: Item, q: str | None = None) 中，item 被识别为什么？

A 路径参数

B 查询参数

C 请求体（因为 Item 是 Pydantic 模型）

D 响应头

✏️ 练习

设计一个课程模型

创建 Course 模型，包含：名称（1~50字符）、学分（1~10 的整数）、教师名、可选的先修课程列表（嵌套 Course 列表）。实现 POST /courses 接口。

main.py

from pydantic import BaseModel, Field
from typing import Optional

class Course(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    credits: int = Field(ge=1, le=10)
    teacher: str
    prerequisites: list["Course"] = []  # 自引用嵌套

@app.post("/courses")
async def create_course(course: Course):
    return {"created": course}

🎯

第三天完成！

你已经掌握了 Pydantic 数据建模、请求体接收和嵌套模型。

DAY 4

响应模型与异常处理

控制输出格式、管理状态码、优雅地处理错误——让 API 更专业。

🎯 目标：掌控输入输出⏱️ 预计：50 分钟

1response_model 控制输出

默认情况下 FastAPI 返回函数返回值的所有字段。用 response_model 可以过滤输出，只暴露你想要的字段——这对安全很重要：

main.py

class UserIn(BaseModel):
    username: str
    password: str       # 输入包含密码
    email: str

class UserOut(BaseModel):  # 输出不包含密码！
    username: str
    email: str

@app.post("/users", response_model=UserOut)
async def create_user(user: UserIn):
    return user  # 接收 UserIn，但只返回 UserOut 的字段
# 响应：{"username": "...", "email": "..."}  ← 没有 password！

为什么需要 response_model？①安全：过滤敏感字段（如密码哈希）；②文档：让 /docs 显示准确的响应结构；③一致性：无论函数内部返回什么，输出格式始终可控。

2状态码

用 status_code 参数指定成功响应的 HTTP 状态码：

main.py

from fastapi import status

@app.post("/items", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return item

# 常用状态码：
# 200 OK            - 请求成功（GET/PUT 默认）
# 201 Created       - 资源创建成功（POST 常用）
# 204 No Content    - 成功但无返回体
# 400 Bad Request   - 客户端请求错误
# 404 Not Found     - 资源不存在
# 422 Unprocessable - 验证失败

3HTTPException 异常处理

当业务逻辑出错时，抛出 HTTPException 返回带状态码和详情的错误响应：

main.py

from fastapi import HTTPException

items = {"foo": "The Foo Wrestlers"}

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(
            status_code=404,
            detail="商品不存在",
            headers={"X-Error": "ItemNotFound"}  # 可选自定义头
        )
    return {"item": items[item_id]}

# 访问 /items/bar -> 404
# {"detail": "商品不存在"}

4自定义异常处理器

定义自己的异常类并注册处理器，实现统一的错误响应格式：

main.py

from fastapi import Request
from fastapi.responses import JSONResponse

class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(UnicornException)
async def unicorn_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"code": 418, "message": f"哎呀，{exc.name} 出了点问题"},
    )

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name)
    return {"name": name}

5响应的其他控制

main.py

from fastapi import Response

# 设置响应头
@app.get("/custom-header")
async def read_header(response: Response):
    response.headers["X-Custom"] = "hello"
    return {"msg": "看响应头"}

# 设置 Cookie
from fastapi import response
@app.post("/login")
async def login(resp: Response):
    resp.set_cookie(key="token", value="abc123", httponly=True)
    return {"msg": "已登录"}

随堂测验

理解检查

为什么创建用户接口应该用 response_model=UserOut 而不是直接返回 UserIn？

A 因为 UserIn 性能差，UserOut 更快

B 为了过滤敏感字段（如密码），不暴露给客户端

C 因为 FastAPI 不允许返回 UserIn 类型

D 为了减少 JSON 序列化时间

✏️ 练习

实现带 404 的查询接口

用一个字典模拟数据库 users = {1:{"name":"张三","age":20}, 2:{"name":"李四","age":22}}。实现 GET /users/{uid}，存在则返回用户，不存在则抛出 404 HTTPException。

main.py

from fastapi import HTTPException

users = {1: {"name": "张三", "age": 20}, 2: {"name": "李四", "age": 22}}

@app.get("/users/{uid}")
async def get_user(uid: int):
    if uid not in users:
        raise HTTPException(status_code=404, detail=f"用户 {uid} 不存在")
    return users[uid]

🎯

第四天完成！

你已经掌握了响应模型、状态码和异常处理。

DAY 5

依赖注入与中间件

FastAPI 的依赖注入系统是它的灵魂——复用逻辑、管理资源、控制访问，全靠它。

🎯 目标：理解 DI 思想⏱️ 预计：55 分钟

1依赖注入是什么？

依赖注入（Dependency Injection, DI） 是一种设计模式：不自己创建所需的对象，而是声明"我需要什么"，由框架帮你注入。在 FastAPI 中用 Depends() 实现。

通俗理解：就像你去餐厅点餐（声明需求），服务员把菜端给你（注入依赖），你不用关心菜怎么做出来的。FastAPI 是服务员，Depends() 是点菜单。

2第一个依赖

main.py

from fastapi import Depends

# 定义依赖：一个普通函数
def common_params(q: str | None = None, skip: int = 0, limit: int = 10):
    return {"q": q, "skip": skip, "limit": limit}

# 在路由中注入
@app.get("/items")
async def read_items(commons: dict = Depends(common_params)):
    return {"message": "商品列表", "params": commons}

@app.get("/users")
async def read_users(commons: dict = Depends(common_params)):
    return {"message": "用户列表", "params": commons}
# 两个接口共享同一套查询参数逻辑，不重复！

3依赖的嵌套与缓存

依赖可以嵌套（依赖中再依赖），且同一个请求内，同一依赖只执行一次（结果被缓存）：

main.py

def query_db(q: str | None = None):
    if q:
        return {"query": q, "results": [f"结果-{q}-1", f"结果-{q}-2"]}
    return {"query": None, "results": []}

def logic_dep(db: dict = Depends(query_db)):
    # 嵌套依赖：logic_dep 依赖 query_db
    return {"db": db, "extra": "附加逻辑"}

@app.get("/search")
async def search(
    dep1: dict = Depends(logic_dep),
    dep2: dict = Depends(query_db),  # query_db 只执行一次！dep2 复用缓存
):
    return {"dep1": dep1, "dep2": dep2}

用 use_cache=False 关闭缓存：如果你需要依赖每次都重新执行，用 Depends(query_db, use_cache=False)。

4依赖管理资源（数据库连接）

依赖非常适合管理需要打开/关闭的资源，配合 yield 语法：

main.py

def get_db():
    db = "打开数据库连接"  # 模拟
    try:
        yield db            # yield 之前的代码 = 请求前执行
        # yield 的值注入到路由函数
    finally:
        print("关闭数据库连接")  # yield 之后 = 请求后执行

@app.get("/data")
async def read_data(db = Depends(get_db)):
    return {"db_status": db}

5全局依赖与路由组

main.py

def verify_token(x_token: str = Header()):
    if x_token != "secret-token":
        raise HTTPException(400, "X-Token 无效")

# 应用级全局依赖：所有路由都执行
app = FastAPI(dependencies=[Depends(verify_token)])

# 路由组（APIRouter）依赖
from fastapi import APIRouter
admin = APIRouter(prefix="/admin", dependencies=[Depends(verify_token)])

@admin.get("/dashboard")
async def dashboard():
    return {"area": "admin"}

app.include_router(admin)

6中间件

中间件在请求到达路由之前、响应返回客户端之前介入，适合日志、CORS、限流等横切关注点：

main.py

import time
from fastapi import Request

@app.middleware("http")
async def timing_middleware(request: Request, call_next):
    start = time.time()
    response = await call_next(request)  # 调用后续处理
    duration = time.time() - start
    response.headers["X-Process-Time"] = f"{duration:.4f}s"
    return response

# CORS 中间件（跨域）
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # 前端地址
    allow_methods=["*"],
    allow_headers=["*"],
)

随堂测验

理解检查

关于 FastAPI 依赖注入的缓存机制，下列说法正确的是？

A 依赖每次调用都会重新执行，不缓存

B 同一请求内，同一依赖只执行一次，结果被缓存复用

C 缓存跨请求有效，第二次请求直接用上次结果

D 只有 async 依赖才缓存，同步依赖不缓存

✏️ 练习

用依赖实现分页参数复用

创建一个 pagination_dep 依赖，接收 page（默认1，≥1）和 size（默认10，1~100），返回 {"offset": (page-1)*size, "limit": size}。在两个接口中使用它。

main.py

from fastapi import Depends, Query

def pagination_dep(
    page: int = Query(default=1, ge=1),
    size: int = Query(default=10, ge=1, le=100),
):
    return {"offset": (page - 1) * size, "limit": size}

@app.get("/articles")
async def list_articles(pg: dict = Depends(pagination_dep)):
    return {"type": "articles", "pagination": pg}

@app.get("/comments")
async def list_comments(pg: dict = Depends(pagination_dep)):
    return {"type": "comments", "pagination": pg}

🎯

第五天完成！

你已经理解了依赖注入思想，能复用逻辑、管理资源和配置中间件。

DAY 6

数据库集成与异步

用 SQLAlchemy 连接数据库，实现完整 CRUD，感受异步编程的威力。

🎯 目标：数据库 CRUD⏱️ 预计：60 分钟

1项目结构

随着项目变大，把代码拆分到多个文件。一个推荐的目录结构：

项目结构

myapi/
├── main.py              # 应用入口
├── database.py          # 数据库连接
├── models.py            # SQLAlchemy 模型
├── schemas.py           # Pydantic 模型（输入输出）
├── crud.py              # 数据库操作函数
└── requirements.txt

2数据库连接层

database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

SQLALCHEMY_DB_URL = "sqlite:///./app.db"  # SQLite 文件数据库

engine = create_engine(
    SQLALCHEMY_DB_URL,
    connect_args={"check_same_thread": False}  # SQLite 专用
)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
Base = declarative_base()

# 依赖：每个请求获取独立 session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

3SQLAlchemy 模型 vs Pydantic 模型

特征	SQLAlchemy 模型	Pydantic 模型
用途	数据库表映射	请求/响应数据验证
定义	继承 Base，用 Column	继承 BaseModel
文件	models.py	schemas.py
关注点	数据怎么存	数据怎么传

models.py

from sqlalchemy import Column, Integer, String, Float
from database import Base

class Product(Base):
    __tablename__ = "products"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), nullable=False)
    price = Column(Float, nullable=False)
    description = Column(String(500), default="")

schemas.py

from pydantic import BaseModel, Field

class ProductCreate(BaseModel):  # 创建时用
    name: str = Field(min_length=1, max_length=100)
    price: float = Field(gt=0)
    description: str = ""

class ProductOut(BaseModel):     # 返回时用（含 id）
    id: int
    name: str
    price: float
    description: str

    class Config:
        from_attributes = True   # 允许从 ORM 对象读取属性

4CRUD 操作

crud.py

from sqlalchemy.orm import Session
import models, schemas

def get_product(db: Session, product_id: int):
    return db.query(models.Product).filter(
        models.Product.id == product_id
    ).first()

def get_products(db: Session, skip: int = 0, limit: int = 20):
    return db.query(models.Product).offset(skip).limit(limit).all()

def create_product(db: Session, product: schemas.ProductCreate):
    db_product = models.Product(**product.model_dump())
    db.add(db_product)
    db.commit()
    db.refresh(db_product)  # 获取自增 id
    return db_product

def delete_product(db: Session, product_id: int):
    product = get_product(db, product_id)
    if product:
        db.delete(product)
        db.commit()
    return product

5路由整合

main.py

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
import models, schemas, crud
from database import engine, get_db

models.Base.metadata.create_all(bind=engine)  # 建表
app = FastAPI()

@app.post("/products", response_model=schemas.ProductOut, status_code=201)
def create_product(product: schemas.ProductCreate, db: Session = Depends(get_db)):
    return crud.create_product(db, product)

@app.get("/products", response_model=list[schemas.ProductOut])
def list_products(skip: int = 0, limit: int = 20, db: Session = Depends(get_db)):
    return crud.get_products(db, skip, limit)

@app.get("/products/{pid}", response_model=schemas.ProductOut)
def read_product(pid: int, db: Session = Depends(get_db)):
    product = crud.get_product(db, pid)
    if not product:
        raise HTTPException(404, "商品不存在")
    return product

@app.delete("/products/{pid}")
def remove_product(pid: int, db: Session = Depends(get_db)):
    product = crud.delete_product(db, pid)
    if not product:
        raise HTTPException(404, "商品不存在")
    return {"deleted": pid}

同步 vs 异步数据库：上面的写法用同步 SQLAlchemy。如果要用异步，需要 aiosqlite/asyncpg + AsyncSession，函数用 async def + await db.execute()。对初学者建议先掌握同步版本。

6异步编程入门

FastAPI 原生支持 async/await。异步的价值在于 I/O 密集场景（数据库、HTTP 请求、文件读写）下不阻塞线程：

main.py

import httpx  # 异步 HTTP 客户端
import asyncio

@app.get("/weather/{city}")
async def get_weather(city: str):
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"https://wttr.in/{city}", params={"format": "j1"}
        )
        data = resp.json()
    return {"city": city, "temp": data["current_condition"][0]["temp_C"]}

# 同时请求多个外部 API（并发）
@app.get("/multi")
async def multi():
    async with httpx.AsyncClient() as client:
        tasks = [client.get(f"https://httpbin.org/delay/{i}") for i in range(3)]
        responses = await asyncio.gather(*tasks)  # 并发执行！
    return {"count": len(responses)}

什么时候用 async def？函数内部有 await 调用时（异步 I/O），必须用 async def。纯计算或同步 I/O 用普通 def 即可（FastAPI 会放到线程池执行，不会阻塞事件循环）。

随堂测验

理解检查

在 FastAPI + SQLAlchemy 项目中，为什么数据库 session 通过 Depends(get_db) 注入而不是在函数内直接创建？

A 因为 Depends 比直接创建性能更好

B 因为 FastAPI 强制要求必须用 Depends

C 为了确保每个请求用独立 session 且请求结束后自动关闭

D 为了能在 /docs 中显示数据库连接信息

✏️ 练习

实现更新接口

在 crud.py 和 main.py 中实现 PUT /products/{pid} 更新接口。创建 ProductUpdate schema（所有字段可选），更新非空字段后返回。

crud.py + main.py

# schemas.py
from typing import Optional
class ProductUpdate(BaseModel):
    name: Optional[str] = None
    price: Optional[float] = None
    description: Optional[str] = None

# crud.py
def update_product(db: Session, pid: int, data: schemas.ProductUpdate):
    product = get_product(db, pid)
    if not product:
        return None
    for key, val in data.model_dump(exclude_unset=True).items():
        setattr(product, key, val)
    db.commit()
    db.refresh(product)
    return product

# main.py
@app.put("/products/{pid}", response_model=schemas.ProductOut)
def update(pid: int, data: schemas.ProductUpdate, db: Session = Depends(get_db)):
    product = crud.update_product(db, pid, data)
    if not product:
        raise HTTPException(404, "商品不存在")
    return product

🎯

第六天完成！

你已经能整合数据库，实现完整 CRUD，并理解了异步编程。

DAY 7

认证、测试与部署

JWT 认证、自动化测试、生产部署——把你的 API 推向实战。

🎯 目标：能上生产⏱️ 预计：65 分钟

1OAuth2 + JWT 认证

JWT（JSON Web Token） 是 API 认证的常用方案：用户登录后获得 token，后续请求携带 token 验证身份。

auth.py

from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_ctx = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 密码哈希
def hash_password(password: str) -> str:
    return pwd_ctx.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_ctx.verify(plain, hashed)

# 生成 token
def create_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

# 当前用户依赖
def get_current_user(token: str = Depends(oauth2_scheme)):
    cred_err = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise cred_err
    except JWTError:
        raise cred_err
    # 这里应该从数据库查用户，简化示例
    return {"username": username}

登录接口签发 token，受保护接口用 Depends(get_current_user)：

main.py

from fastapi.security import OAuth2PasswordRequestForm
from auth import create_token, get_current_user, verify_password, hash_password

# 模拟用户库（实际用数据库）
fake_users = {"alice": {"username": "alice", "hashed": hash_password("secret")}}

@app.post("/token")
async def login(form: OAuth2PasswordRequestForm = Depends()):
    user = fake_users.get(form.username)
    if not user or not verify_password(form.password, user["hashed"]):
        raise HTTPException(401, "用户名或密码错误")
    token = create_token({"sub": user["username"]})
    return {"access_token": token, "token_type": "bearer"}

@app.get("/me")
async def me(current = Depends(get_current_user)):
    return current

2自动化测试

FastAPI 提供 TestClient，基于 httpx，无需真正启动服务器即可测试：

test_main.py

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello, FastAPI!", "status": "running"}

def test_read_item():
    response = client.get("/items/42", params={"q": "test"})
    assert response.status_code == 200
    assert response.json()["item_id"] == 42
    assert response.json()["q"] == "test"

def test_invalid_item_id():
    response = client.get("/items/not-a-number")
    assert response.status_code == 422  # 验证失败

def test_create_item():
    response = client.post("/items/", json={
        "name": "键盘", "price": 299.0, "is_offer": True, "tags": ["外设"]
    })
    assert response.status_code == 200
    assert response.json()["received"]["name"] == "键盘"

运行测试

pip install pytest
pytest test_main.py -v

# 输出示例：
# test_main.py::test_read_root PASSED
# test_main.py::test_read_item PASSED
# test_main.py::test_invalid_item_id PASSED
# test_main.py::test_create_item PASSED

3环境变量与配置

config.py

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "My API"
    database_url: str = "sqlite:///./app.db"
    secret_key: str = "dev-secret"
    debug: bool = True

    class Config:
        env_file = ".env"

settings = Settings()
# 读取 .env 文件或环境变量，类型安全

.env

APP_NAME=生产 API
DATABASE_URL=postgresql://user:pass@localhost:5432/mydb
SECRET_KEY=a-very-long-random-string
DEBUG=False

4部署到生产

部署命令

# 用 Gunicorn 管理 Uvicorn worker（生产推荐）
pip install gunicorn

gunicorn main:app \
  -w 4 \                    # 4 个 worker 进程
  -k uvicorn.workers.UvicornWorker \
  -b 0.0.0.0:8000

# Docker 部署
# Dockerfile:
#   FROM python:3.12-slim
#   WORKDIR /app
#   COPY requirements.txt .
#   RUN pip install --no-cache-dir -r requirements.txt
#   COPY . .
#   CMD ["gunicorn", "main:app", "-w", "4", \
#        "-k", "uvicorn.workers.UvicornWorker", \
#        "-b", "0.0.0.0:8000"]

场景	命令	说明
开发	`uvicorn main:app --reload`	热重载，单进程
生产	`gunicorn ... -k UvicornWorker`	多 worker，稳定
Docker	`docker build -t myapi . && docker run -p 8000:8000 myapi`	容器化部署
反代	Nginx → Uvicorn	处理 TLS、静态文件、负载均衡

5进阶路线图

🔌

WebSocket

实时双向通信，适合聊天、推送。

📄

后台任务

BackgroundTasks 发邮件、Celery 做重活。

🔍

APIRouter

大项目拆分路由，微服务化。

📊

分页/缓存

Redis 缓存、游标分页。

🐳

Docker Compose

编排 API + DB + redis 多容器。

📈

监控

Prometheus + Grafana 指标监控。

🎉

恭喜你完成了七日速成！

从 Hello World 到认证、测试、部署——你已经具备了用 FastAPI 构建生产级 API 的基础知识。接下来最好的学习方式就是：动手做一个项目。

随堂测验

理解检查

关于 JWT token 认证，下列说法正确的是？

A Token 存在服务端，每次请求查服务端验证

B Token 包含签名信息，服务端用密钥验证真伪，无需查库

C Token 一旦签发永不过期

D JWT 只能用 HTTPS 传输，HTTP 下不可用

✏️ 练习

写一个测试用例

为 POST /items/ 接口写一个测试：验证当 price 传负数时返回 422 状态码。

test_main.py

def test_create_item_invalid_price():
    response = client.post("/items/", json={
        "name": "测试商品", "price": -10.0
    })
    assert response.status_code == 422
    # 可进一步检查错误详情
    detail = response.json()["detail"]
    assert any(d["loc"] == ["body", "price"] for d in detail)

🏆

全部课程完成！

你已走完 FastAPI 速成之旅。标记完成，记录你的成就！