1. FastAPI 是什么？ #

FastAPI 是一个现代、快速的 Python Web 框架，专门用于构建 API（应用程序接口）。它基于 Python 类型提示，让你用简单的代码就能创建高性能的 Web API。

1.1 为什么需要 FastAPI？ #

想象一下，如果不用框架，你要这样创建一个简单的 API：

# 传统方式：需要手动处理 HTTP 请求、解析参数、验证数据等（非常复杂）
# 这里省略了复杂的实现代码...

使用 FastAPI 后，你可以这样写：

# FastAPI 方式：简单、直观、自动处理很多细节
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_root():
    return {"message": "Hello FastAPI"}

1.2 FastAPI 的核心优势 #

1. 代码简洁：用 Python 类型提示自动处理数据验证
2. 性能优秀：是目前最快的 Python Web 框架之一
3. 自动文档：自动生成交互式 API 文档，无需手动编写
4. 类型安全：利用 Python 类型提示，减少错误
5. 易于学习：基于 Python 标准特性，学习曲线平缓

2. 前置知识 #

在学习 FastAPI 之前，你需要了解以下基础知识。

2.1 什么是 API？ #

API（Application Programming Interface，应用程序接口） 是不同软件之间通信的桥梁。在 Web 开发中，API 通常指 RESTful API，它使用 HTTP 协议进行通信。

简单理解：

• 客户端（如浏览器、手机 App）发送 HTTP 请求
• 服务器（你的 FastAPI 应用）处理请求并返回响应
• 数据通常以 JSON 格式传输

2.2 HTTP 基础知识 #

• 请求方法：

  • GET：获取数据（如查看用户信息）
  • POST：创建数据（如创建新用户）
  • PUT：更新数据（如修改用户信息）
  • DELETE：删除数据（如删除用户）

• URL 路径：如 /users/123，表示访问 ID 为 123 的用户

• 查询参数：如 /users?skip=0&limit=10，skip 和 limit 是查询参数

• 请求体：POST/PUT 请求可以携带数据（通常是 JSON 格式）

2.3 Python 类型提示 #

FastAPI 大量使用 Python 类型提示，你需要了解：

# 基本类型提示
name: str          # 字符串类型
age: int           # 整数类型
price: float       # 浮点数类型
is_active: bool    # 布尔类型

# 可选类型（可以为 None）
from typing import Optional
description: Optional[str] = None  # 推荐使用 Optional，兼容所有 Python 版本

# 列表类型
items: list[str]   # Python 3.9+
# 或
from typing import List
items: List[str]   # Python 3.9 及以下

如果你对类型提示不熟悉，建议先学习 Python 基础教程。

2.4 async/await 基础 #

FastAPI 支持异步编程，你需要了解：

• async def：定义异步函数
• await：等待异步操作完成

如果你不熟悉异步编程，可以先使用普通函数（def），FastAPI 也支持。

3. 安装与环境准备 #

3.1 安装 FastAPI #

使用 pip 安装：

pip install fastapi
pip install uvicorn[standard]

说明：

• fastapi：FastAPI 框架本身
• uvicorn：ASGI 服务器，用于运行 FastAPI 应用

3.2 验证安装 #

创建一个简单的测试文件 test_install.py：

# 导入 FastAPI
from fastapi import FastAPI

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

# 定义根路径的处理函数
@app.get("/")
async def read_root():
    # 返回一个字典（FastAPI 会自动转换为 JSON）
    return {"message": "FastAPI 安装成功！"}

# 运行应用（使用 uvicorn）
if __name__ == "__main__":
    import uvicorn
    # 运行服务器，监听 8000 端口
    uvicorn.run(app, host="0.0.0.0", port=8000)

运行测试：

python test_install.py

然后在浏览器访问 http://localhost:8000，应该能看到 {"message": "FastAPI 安装成功！"}。

4. 第一个完整示例：Hello World #

让我们创建一个最简单的 FastAPI 应用，理解基本结构。

# 导入 FastAPI 类
from fastapi import FastAPI

# 创建 FastAPI 应用实例
# app 是应用的核心对象，所有路由都注册到它上面
app = FastAPI()

# 定义路由：处理 GET 请求到根路径 "/"
# @app.get("/") 是装饰器，表示这是一个处理 GET 请求的函数
@app.get("/")
async def read_root():
    # 返回一个字典，FastAPI 会自动转换为 JSON 响应
    return {"message": "Hello World"}

# 定义另一个路由：处理 GET 请求到 "/hello"
@app.get("/hello")
async def say_hello():
    # 返回问候语
    return {"message": "你好，FastAPI！"}

# 运行应用
if __name__ == "__main__":
    import uvicorn
    # uvicorn.run() 启动服务器
    # host="0.0.0.0" 表示监听所有网络接口
    # port=8000 表示使用 8000 端口
    uvicorn.run(app, host="0.0.0.0", port=8000)

运行方法：

1. 将代码保存为 hello.py
2. 运行 python hello.py
3. 在浏览器访问 http://localhost:8000 或 http://localhost:8000/hello

或者使用命令行运行：

uvicorn hello:app --reload

其中 hello 是文件名（不含 .py），app 是 FastAPI 实例的变量名。

5. 路径参数：从 URL 中获取数据 #

路径参数是 URL 路径中的一部分，用于传递数据。例如，/users/123 中的 123 就是路径参数。

# 导入 FastAPI
from fastapi import FastAPI

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

# 定义路由，{user_id} 是路径参数
# FastAPI 会自动从 URL 中提取 user_id 并转换为 int 类型
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # user_id 已经是整数类型，可以直接使用
    return {"user_id": user_id, "message": f"获取用户 {user_id} 的信息"}

# 路径参数也可以是字符串类型
@app.get("/items/{item_name}")
async def get_item(item_name: str):
    # item_name 是字符串类型
    return {"item_name": item_name, "message": f"获取商品 {item_name}"}

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

测试方法：

• 访问 http://localhost:8000/users/123，会返回 {"user_id": 123, "message": "获取用户 123 的信息"}
• 访问 http://localhost:8000/items/apple，会返回 {"item_name": "apple", "message": "获取商品 apple"}

注意：如果访问 /users/abc（传入非数字），FastAPI 会自动返回错误，因为 user_id 被声明为 int 类型。

6. 查询参数：从 URL 问号后获取数据 #

查询参数是 URL 中 ? 后面的参数，如 /items?skip=0&limit=10 中的 skip=0 和 limit=10。

# 导入 FastAPI 和 Optional
from fastapi import FastAPI
from typing import Optional

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

# 定义路由，函数参数中不是路径参数的，会被视为查询参数
# skip 和 limit 是查询参数，有默认值
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    # skip 和 limit 从 URL 查询参数中获取
    # 例如：/items/?skip=20&limit=5
    return {
        "skip": skip,
        "limit": limit,
        "message": f"跳过 {skip} 条，返回 {limit} 条"
    }

# 查询参数也可以是可选的（使用 None 作为默认值）
@app.get("/search/")
async def search_items(q: Optional[str] = None):
    # q 是可选参数，如果 URL 中没有提供，则为 None
    if q:
        return {"query": q, "message": f"搜索关键词：{q}"}
    else:
        return {"message": "未提供搜索关键词"}

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

测试方法：

• 访问 http://localhost:8000/items/，会使用默认值 skip=0, limit=10
• 访问 http://localhost:8000/items/?skip=20&limit=5，会使用提供的值
• 访问 http://localhost:8000/search/?q=python，会返回搜索结果

7. 请求体：接收 JSON 数据 #

请求体是 POST/PUT 请求中发送的数据（通常是 JSON 格式）。FastAPI 使用 Pydantic 模型来定义和验证请求体。

7.1 使用 Pydantic 模型定义请求体 #

# 导入 FastAPI、Pydantic 和 Optional
from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional

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

# 定义 Pydantic 模型：用于描述请求体的数据结构
# BaseModel 是 Pydantic 的基类，提供数据验证功能
class Item(BaseModel):
    # name 是必填字段，类型为 str
    name: str
    # price 是必填字段，类型为 float
    price: float
    # description 是可选字段，默认值为 None
    description: Optional[str] = None
    # is_available 是可选字段，默认值为 True
    is_available: bool = True

# 定义 POST 路由：创建新商品
@app.post("/items/")
async def create_item(item: Item):
    # item 是 Item 类型的对象，FastAPI 会自动从请求体中解析并验证
    # 如果数据不符合要求（如缺少必填字段、类型错误），会自动返回错误
    return {
        "message": "商品创建成功",
        "item_name": item.name,
        "item_price": item.price,
        "item_description": item.description,
        "item_available": item.is_available
    }

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

测试方法（使用 curl 或 Postman）：

curl -X POST "http://localhost:8000/items/" \
  -H "Content-Type: application/json" \
  -d '{"name": "苹果", "price": 5.5, "description": "新鲜苹果"}'

或者在浏览器中访问 http://localhost:8000/docs，使用自动生成的交互式文档进行测试。

7.2 组合使用路径参数、查询参数和请求体 #

# 导入 FastAPI、Pydantic 和 Optional
from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional

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

# 定义商品模型
class Item(BaseModel):
    name: str
    price: float
    description: Optional[str] = None

# 定义路由：可以同时使用路径参数、查询参数和请求体
@app.put("/items/{item_id}")
async def update_item(
    item_id: int,           # 路径参数
    item: Item,              # 请求体
    q: Optional[str] = None    # 查询参数（可选）
):
    # 返回所有参数
    result = {
        "item_id": item_id,
        "item_name": item.name,
        "item_price": item.price,
        "item_description": item.description
    }
    # 如果提供了查询参数 q，添加到结果中
    if q:
        result["query"] = q
    return result

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

8. 响应模型：控制返回数据的格式 #

响应模型用于定义 API 返回数据的结构，FastAPI 会自动验证返回的数据是否符合模型定义。

# 导入 FastAPI 和 Pydantic
from fastapi import FastAPI
from pydantic import BaseModel

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

# 定义请求模型（用于接收数据）
class ItemCreate(BaseModel):
    name: str
    price: float
    description: Optional[str] = None

# 定义响应模型（用于返回数据）
# 响应模型可以隐藏某些字段，或添加额外字段
class ItemResponse(BaseModel):
    id: int
    name: str
    price: float
    # description 字段在响应中不包含（即使请求中有）

# 模拟数据库（实际应用中应该使用真实数据库）
items_db = []
next_id = 1

# 定义 POST 路由，使用 response_model 指定响应格式
@app.post("/items/", response_model=ItemResponse)
async def create_item(item: ItemCreate):
    global next_id
    # 创建新商品
    new_item = {
        "id": next_id,
        "name": item.name,
        "price": item.price
    }
    # 保存到"数据库"
    items_db.append(new_item)
    next_id += 1
    # 返回的数据会自动按照 ItemResponse 模型格式化
    return new_item

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

9. HTTP 方法：GET、POST、PUT、DELETE #

FastAPI 支持所有常见的 HTTP 方法，让我们创建一个完整的 CRUD（增删改查）示例。

# 导入 FastAPI 和 Pydantic
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List

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

# 定义商品模型
class Item(BaseModel):
    name: str
    price: float
    description: Optional[str] = None

# 定义带 ID 的商品模型（用于响应）
class ItemWithId(Item):
    id: int

# 模拟数据库（实际应用中应该使用真实数据库）
items_db: List[ItemWithId] = []
next_id = 1

# GET：获取所有商品
@app.get("/items/", response_model=List[ItemWithId])
async def get_all_items():
    # 返回所有商品
    return items_db

# GET：根据 ID 获取单个商品
@app.get("/items/{item_id}", response_model=ItemWithId)
async def get_item(item_id: int):
    # 查找商品
    for item in items_db:
        if item.id == item_id:
            return item
    # 如果没找到，抛出 404 错误
    raise HTTPException(status_code=404, detail="商品不存在")

# POST：创建新商品
@app.post("/items/", response_model=ItemWithId)
async def create_item(item: Item):
    global next_id
    # 创建新商品对象
    new_item = ItemWithId(id=next_id, **item.dict())
    # 保存到"数据库"
    items_db.append(new_item)
    next_id += 1
    return new_item

# PUT：更新商品
@app.put("/items/{item_id}", response_model=ItemWithId)
async def update_item(item_id: int, item: Item):
    # 查找商品
    for i, existing_item in enumerate(items_db):
        if existing_item.id == item_id:
            # 更新商品信息
            updated_item = ItemWithId(id=item_id, **item.dict())
            items_db[i] = updated_item
            return updated_item
    # 如果没找到，抛出 404 错误
    raise HTTPException(status_code=404, detail="商品不存在")

# DELETE：删除商品
@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    # 查找并删除商品
    for i, item in enumerate(items_db):
        if item.id == item_id:
            # 从列表中删除
            deleted_item = items_db.pop(i)
            return {"message": "删除成功", "deleted_item": deleted_item}
    # 如果没找到，抛出 404 错误
    raise HTTPException(status_code=404, detail="商品不存在")

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

测试方法（使用 curl）：

# 创建商品
curl -X POST "http://localhost:8000/items/" \
  -H "Content-Type: application/json" \
  -d '{"name": "苹果", "price": 5.5, "description": "新鲜苹果"}'

# 获取所有商品
curl "http://localhost:8000/items/"

# 获取单个商品
curl "http://localhost:8000/items/1"

# 更新商品
curl -X PUT "http://localhost:8000/items/1" \
  -H "Content-Type: application/json" \
  -d '{"name": "红苹果", "price": 6.0, "description": "优质红苹果"}'

# 删除商品
curl -X DELETE "http://localhost:8000/items/1"

10. APIRouter：组织和管理路由 #

当你的应用变得越来越大时，把所有路由都写在一个文件中会变得难以维护。APIRouter 可以帮助你将路由组织到不同的模块中，让代码更清晰、更易管理。

10.1 什么是 APIRouter？ #

APIRouter 是 FastAPI 提供的路由组织工具，它允许你：

• 将相关的路由分组到不同的模块
• 为路由组添加统一的前缀（如 /api/v1）
• 为路由组添加标签（用于 API 文档分类）
• 在多个文件中定义路由，然后在主应用中统一包含

简单理解：

• 没有 APIRouter：所有路由都在一个文件中，像把所有文件都放在一个文件夹里
• 使用 APIRouter：将路由分组到不同模块，像把文件分类放到不同文件夹里

10.2 为什么需要 APIRouter？ #

问题场景：假设你的应用有用户管理、商品管理、订单管理等功能，如果所有路由都写在一个文件中：

# 不好的做法：所有路由都在一个文件中
app = FastAPI()

@app.get("/users/")
async def get_users(): ...

@app.post("/users/")
async def create_user(): ...

@app.get("/items/")
async def get_items(): ...

@app.post("/items/")
async def create_item(): ...

@app.get("/orders/")
async def get_orders(): ...

@app.post("/orders/")
async def create_order(): ...

问题：

• 文件会变得很长，难以维护
• 不同功能的代码混在一起，难以查找
• 多人协作时容易产生冲突

解决方案：使用 APIRouter 将不同功能的路由分组：

# 好的做法：使用 APIRouter 分组
# routers/users.py
router = APIRouter()

@router.get("/")
async def get_users(): ...

@router.post("/")
async def create_user(): ...

# routers/items.py
router = APIRouter()

@router.get("/")
async def get_items(): ...

@router.post("/")
async def create_item(): ...

10.3 基本使用：创建和使用 APIRouter #

# 导入 FastAPI 和 APIRouter
from fastapi import FastAPI, APIRouter

# 创建主应用
app = FastAPI()

# 创建一个路由器实例
# APIRouter 的使用方式和 FastAPI 应用类似
router = APIRouter()

# 在路由器上定义路由（使用 @router 而不是 @app）
@router.get("/users/")
async def get_users():
    # 返回用户列表
    return {"users": ["张三", "李四", "王五"]}

@router.get("/users/{user_id}")
async def get_user(user_id: int):
    # 返回单个用户信息
    return {"user_id": user_id, "name": f"用户{user_id}"}

# 将路由器包含到主应用中
# prefix 参数为所有路由添加统一前缀
# tags 参数为路由添加标签（用于 API 文档分类）
app.include_router(router, prefix="/api", tags=["用户"])

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

说明：

• APIRouter()：创建一个路由器实例
• `@router.get()：在路由器上定义路由（而不是@app.get()`）
• app.include_router()：将路由器包含到主应用中
• prefix="/api"：为所有路由添加 /api 前缀（实际路径是 /api/users/）
• tags=["用户"]：为路由添加标签，在 API 文档中会分组显示

访问路径：

• http://localhost:8000/api/users/：获取用户列表
• http://localhost:8000/api/users/1：获取用户 1 的信息

10.4 多个路由器：组织不同功能模块 #

在实际项目中，你会将不同功能的路由放到不同的文件中。

步骤 1：创建用户路由模块

# routers/users.py
# 导入 APIRouter 和必要的模块
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import List

# 创建路由器实例
# prefix 可以在创建路由器时指定，也可以在 include_router 时指定
router = APIRouter(prefix="/users", tags=["用户"])

# 定义用户模型
class User(BaseModel):
    id: int
    name: str
    email: str

# 模拟用户数据库
users_db: List[User] = [
    User(id=1, name="张三", email="zhangsan@example.com"),
    User(id=2, name="李四", email="lisi@example.com")
]

# 定义路由：获取所有用户
@router.get("/", response_model=List[User])
async def get_users():
    # 返回所有用户
    return users_db

# 定义路由：根据 ID 获取用户
@router.get("/{user_id}", response_model=User)
async def get_user(user_id: int):
    # 查找用户
    for user in users_db:
        if user.id == user_id:
            return user
    # 如果没找到，抛出 404 错误
    raise HTTPException(status_code=404, detail="用户不存在")

# 定义路由：创建新用户
@router.post("/", response_model=User)
async def create_user(user: User):
    # 检查用户 ID 是否已存在
    for existing_user in users_db:
        if existing_user.id == user.id:
            raise HTTPException(status_code=400, detail="用户 ID 已存在")
    # 添加新用户
    users_db.append(user)
    return user

步骤 2：创建商品路由模块

# routers/items.py
# 导入 APIRouter 和必要的模块
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import List, Optional

# 创建路由器实例
router = APIRouter(prefix="/items", tags=["商品"])

# 定义商品模型
class Item(BaseModel):
    id: int
    name: str
    price: float
    description: Optional[str] = None

# 模拟商品数据库
items_db: List[Item] = [
    Item(id=1, name="苹果", price=5.5, description="新鲜苹果"),
    Item(id=2, name="香蕉", price=3.0, description="优质香蕉")
]

# 定义路由：获取所有商品
@router.get("/", response_model=List[Item])
async def get_items():
    # 返回所有商品
    return items_db

# 定义路由：根据 ID 获取商品
@router.get("/{item_id}", response_model=Item)
async def get_item(item_id: int):
    # 查找商品
    for item in items_db:
        if item.id == item_id:
    return item
    # 如果没找到，抛出 404 错误
    raise HTTPException(status_code=404, detail="商品不存在")

# 定义路由：创建新商品
@router.post("/", response_model=Item)
async def create_item(item: Item):
    # 检查商品 ID 是否已存在
    for existing_item in items_db:
        if existing_item.id == item.id:
            raise HTTPException(status_code=400, detail="商品 ID 已存在")
    # 添加新商品
    items_db.append(item)
    return item

步骤 3：在主应用中包含所有路由器

# main.py
# 导入 FastAPI
from fastapi import FastAPI

# 导入路由器模块
from routers import users, items

# 创建主应用实例
app = FastAPI(
    title="我的 API",
    description="使用 APIRouter 组织的 API",
    version="1.0.0"
)

# 包含用户路由器
# 注意：如果路由器中已经指定了 prefix，这里就不需要再指定了
app.include_router(users.router)

# 包含商品路由器
app.include_router(items.router)

# 定义根路径
@app.get("/")
async def read_root():
    # 返回欢迎信息
    return {
        "message": "欢迎使用 API",
        "docs": "/docs",
        "users": "/users/",
        "items": "/items/"
    }

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

项目结构：

项目目录/
├── main.py           # 主应用文件
├── routers/          # 路由模块目录
│   ├── __init__.py   # 空文件，使 routers 成为 Python 包
│   ├── users.py      # 用户路由
│   └── items.py      # 商品路由

访问路径：

• http://localhost:8000/：根路径
• http://localhost:8000/users/：获取所有用户
• http://localhost:8000/users/1：获取用户 1
• http://localhost:8000/items/：获取所有商品
• http://localhost:8000/items/1：获取商品 1
• http://localhost:8000/docs：查看 API 文档（会按标签分组显示）

10.5 路由器的前缀和标签 #

你可以在创建路由器时指定前缀和标签，也可以在包含路由器时指定。

方式 1：在创建路由器时指定

# routers/users.py
from fastapi import APIRouter

# 在创建路由器时指定前缀和标签
router = APIRouter(
    prefix="/api/v1/users",  # 所有路由都会添加这个前缀
    tags=["用户管理"]        # 在 API 文档中会显示这个标签
)

@router.get("/")
async def get_users():
    return {"users": []}

方式 2：在包含路由器时指定

# main.py
from fastapi import FastAPI
from routers import users

app = FastAPI()

# 在包含路由器时指定前缀和标签
# 如果路由器中已经指定了 prefix，这里的 prefix 会追加
app.include_router(
    users.router,
    prefix="/api/v1",  # 追加前缀
    tags=["用户"]       # 添加标签
)

推荐做法：在创建路由器时指定 prefix 和 tags，这样更清晰，每个路由器自己管理自己的配置。

10.6 完整示例：多模块项目结构 #

这是一个完整的示例，展示如何使用 APIRouter 组织一个多模块的项目。

项目结构：

myapi/
├── main.py
├── routers/
│   ├── __init__.py
│   ├── users.py
│   └── items.py
└── models.py

models.py（共享的数据模型）：

# models.py
# 导入 Pydantic
from pydantic import BaseModel
from typing import Optional

# 定义用户模型
class User(BaseModel):
    id: int
    name: str
    email: str

# 定义用户创建模型（不需要 ID）
class UserCreate(BaseModel):
    name: str
    email: str

# 定义商品模型
class Item(BaseModel):
    id: int
    name: str
    price: float
    description: Optional[str] = None

# 定义商品创建模型（不需要 ID）
class ItemCreate(BaseModel):
    name: str
    price: float
    description: Optional[str] = None

routers/users.py（用户路由）：

# routers/users.py
# 导入必要的模块
from fastapi import APIRouter, HTTPException
from typing import List
from models import User, UserCreate

# 创建路由器，指定前缀和标签
router = APIRouter(prefix="/users", tags=["用户"])

# 模拟用户数据库
users_db: List[User] = []
next_id = 1

# 定义路由：获取所有用户
@router.get("/", response_model=List[User])
async def get_users():
    # 返回所有用户
    return users_db

# 定义路由：根据 ID 获取用户
@router.get("/{user_id}", response_model=User)
async def get_user(user_id: int):
    # 查找用户
    for user in users_db:
        if user.id == user_id:
            return user
    # 如果没找到，抛出 404 错误
    raise HTTPException(status_code=404, detail="用户不存在")

# 定义路由：创建新用户
@router.post("/", response_model=User)
async def create_user(user: UserCreate):
    global next_id
    # 创建新用户对象
    new_user = User(id=next_id, **user.dict())
    # 添加到数据库
    users_db.append(new_user)
    next_id += 1
    return new_user

routers/items.py（商品路由）：

# routers/items.py
# 导入必要的模块
from fastapi import APIRouter, HTTPException
from typing import List
from models import Item, ItemCreate

# 创建路由器，指定前缀和标签
router = APIRouter(prefix="/items", tags=["商品"])

# 模拟商品数据库
items_db: List[Item] = []
next_id = 1

# 定义路由：获取所有商品
@router.get("/", response_model=List[Item])
async def get_items():
    # 返回所有商品
    return items_db

# 定义路由：根据 ID 获取商品
@router.get("/{item_id}", response_model=Item)
async def get_item(item_id: int):
    # 查找商品
    for item in items_db:
        if item.id == item_id:
            return item
    # 如果没找到，抛出 404 错误
    raise HTTPException(status_code=404, detail="商品不存在")

# 定义路由：创建新商品
@router.post("/", response_model=Item)
async def create_item(item: ItemCreate):
    global next_id
    # 创建新商品对象
    new_item = Item(id=next_id, **item.dict())
    # 添加到数据库
    items_db.append(new_item)
    next_id += 1
    return new_item

main.py（主应用）：

# main.py
# 导入 FastAPI
from fastapi import FastAPI

# 导入路由器模块
from routers import users, items

# 创建主应用实例
app = FastAPI(
    title="多模块 API 示例",
    description="使用 APIRouter 组织的多模块 API",
    version="1.0.0"
)

# 包含用户路由器
app.include_router(users.router)

# 包含商品路由器
app.include_router(items.router)

# 定义根路径
@app.get("/")
async def read_root():
    # 返回 API 信息
    return {
        "message": "欢迎使用多模块 API",
        "docs": "/docs",
        "endpoints": {
            "users": "/users/",
            "items": "/items/"
        }
    }

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

运行和测试：

# 运行应用
python main.py

# 或者使用 uvicorn
uvicorn main:app --reload

访问路径：

• http://localhost:8000/：根路径
• http://localhost:8000/users/：获取所有用户
• http://localhost:8000/users/1：获取用户 1
• http://localhost:8000/items/：获取所有商品
• http://localhost:8000/items/1：获取商品 1
• http://localhost:8000/docs：查看 API 文档（会按"用户"和"商品"标签分组）

10.7 APIRouter 的优势 #

使用 APIRouter 的好处：

1. 代码组织清晰：不同功能的路由分离到不同文件，易于查找和维护
2. 易于扩展：添加新功能只需创建新的路由器文件
3. 团队协作友好：不同开发者可以负责不同的路由器模块，减少冲突
4. 统一管理：可以为路由组统一添加前缀、标签、依赖等
5. API 文档友好：标签功能让 API 文档更清晰，按功能分组显示

10.8 常见使用场景 #

场景 1：API 版本管理

# routers/v1/users.py
router = APIRouter(prefix="/api/v1/users", tags=["用户 v1"])

# routers/v2/users.py
router = APIRouter(prefix="/api/v2/users", tags=["用户 v2"])

# main.py
app.include_router(v1.users.router)
app.include_router(v2.users.router)

场景 2：功能模块分离

# routers/auth.py - 认证相关
router = APIRouter(prefix="/auth", tags=["认证"])

# routers/users.py - 用户管理
router = APIRouter(prefix="/users", tags=["用户"])

# routers/admin.py - 管理员功能
router = APIRouter(prefix="/admin", tags=["管理员"])

场景 3：不同权限的路由

# routers/public.py - 公开路由
router = APIRouter(prefix="/public", tags=["公开"])

# routers/private.py - 需要认证的路由
router = APIRouter(prefix="/private", tags=["私有"])

11. Depends：依赖注入系统 #

依赖注入（Dependency Injection） 是 FastAPI 的一个强大特性，它允许你声明和处理依赖关系，如数据库连接、认证、权限检查等。使用 Depends 可以让代码更清晰、更易测试、更易维护。

11.1 什么是 Depends？ #

Depends 是 FastAPI 提供的依赖注入装饰器，它允许你：

• 在路由函数中声明依赖
• 自动处理依赖的创建和注入
• 复用代码逻辑（如认证、数据库连接等）
• 让代码更模块化和可测试

简单理解：

• 没有 Depends：每个路由函数都要自己处理认证、数据库连接等
• 使用 Depends：将这些逻辑提取为依赖函数，路由函数只需声明需要什么依赖

11.2 为什么需要 Depends？ #

问题场景：假设你有多个路由都需要进行用户认证：

# 不好的做法：每个路由都要自己处理认证
@app.get("/users/me")
async def get_current_user():
    # 从请求头获取 token
    token = request.headers.get("Authorization")
    if not token:
        raise HTTPException(status_code=401, detail="未授权")
    # 验证 token
    user = verify_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="无效的 token")
    return user

@app.get("/users/profile")
async def get_profile():
    # 重复的认证逻辑
    token = request.headers.get("Authorization")
    if not token:
        raise HTTPException(status_code=401, detail="未授权")
    user = verify_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="无效的 token")
    # 获取用户资料
    return get_user_profile(user)

问题：

• 代码重复：认证逻辑在每个路由中都要写一遍
• 难以维护：如果要修改认证逻辑，需要修改多个地方
• 难以测试：认证逻辑和业务逻辑混在一起

解决方案：使用 Depends 将认证逻辑提取为依赖：

# 好的做法：使用 Depends 提取认证逻辑
async def get_current_user(token: str = Header(...)):
    user = verify_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="无效的 token")
    return user

@app.get("/users/me")
async def get_me(current_user: User = Depends(get_current_user)):
    return current_user

@app.get("/users/profile")
async def get_profile(current_user: User = Depends(get_current_user)):
    return get_user_profile(current_user)

11.3 基本使用：函数依赖 #

最简单的依赖是函数依赖，将可复用的逻辑提取为函数。

# 导入 FastAPI 和 Depends
from fastapi import FastAPI, Depends, Header, HTTPException
from typing import Optional

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

# 模拟用户数据库
users_db = {
    "token123": {"id": 1, "name": "张三", "email": "zhangsan@example.com"},
    "token456": {"id": 2, "name": "李四", "email": "lisi@example.com"}
}

# 定义依赖函数：获取当前用户
# 这个函数会从请求头中获取 token，并验证用户
async def get_current_user(authorization: Optional[str] = Header(None)):
    # 检查是否提供了 token
    if not authorization:
        # 抛出 401 错误（未授权）
        raise HTTPException(
            status_code=401,
            detail="缺少认证信息，请提供 Authorization 头"
        )

    # 从 "Bearer token" 格式中提取 token
    # 例如：Authorization: Bearer token123
    if authorization.startswith("Bearer "):
        token = authorization[7:]  # 去掉 "Bearer " 前缀
    else:
        token = authorization

    # 验证 token 并获取用户信息
    if token not in users_db:
        # 抛出 401 错误（无效的 token）
        raise HTTPException(
            status_code=401,
            detail="无效的认证 token"
        )

    # 返回用户信息
    return users_db[token]

# 定义路由：获取当前用户信息
# 使用 Depends(get_current_user) 声明依赖
# FastAPI 会自动调用 get_current_user 函数，并将结果注入到 current_user 参数
@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(get_current_user)):
    # current_user 已经是经过验证的用户信息
    return {
        "message": "当前用户信息",
        "user": current_user
    }

# 定义路由：获取用户资料
# 同样使用 get_current_user 依赖
@app.get("/users/profile")
async def get_user_profile(current_user: dict = Depends(get_current_user)):
    # 使用已验证的用户信息
    return {
        "message": "用户资料",
        "user_id": current_user["id"],
        "name": current_user["name"],
        "email": current_user["email"]
    }

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

测试方法（使用 curl 或 Postman）：

# 不带认证信息（会返回 401 错误）
curl "http://localhost:8000/users/me"

# 带认证信息
curl "http://localhost:8000/users/me" \
  -H "Authorization: Bearer token123"

# 访问用户资料
curl "http://localhost:8000/users/profile" \
  -H "Authorization: Bearer token123"

11.4 查询参数依赖：从 URL 参数获取数据 #

依赖函数也可以从查询参数中获取数据。

# 导入 FastAPI 和 Depends
from fastapi import FastAPI, Depends, Query
from typing import Optional

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

# 定义依赖函数：分页参数
# 这个函数会从查询参数中获取分页信息
async def get_pagination(
    skip: int = Query(0, ge=0, description="跳过的记录数"),
    limit: int = Query(10, ge=1, le=100, description="返回的记录数（1-100）")
):
    # 返回分页信息
    return {
        "skip": skip,
        "limit": limit
    }

# 定义路由：获取商品列表
# 使用 Depends(get_pagination) 声明依赖
@app.get("/items/")
async def get_items(pagination: dict = Depends(get_pagination)):
    # pagination 包含 skip 和 limit 信息
    skip = pagination["skip"]
    limit = pagination["limit"]

    # 模拟商品数据
    all_items = [
        {"id": i, "name": f"商品{i}"} 
        for i in range(1, 101)
    ]

    # 应用分页
    items = all_items[skip:skip + limit]

    return {
        "total": len(all_items),
        "skip": skip,
        "limit": limit,
        "items": items
    }

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

测试方法：

# 使用默认分页参数
curl "http://localhost:8000/items/"

# 指定分页参数
curl "http://localhost:8000/items/?skip=10&limit=20"

11.5 类依赖：使用类作为依赖 #

除了函数，你也可以使用类作为依赖。

# 导入 FastAPI 和 Depends
from fastapi import FastAPI, Depends, Header, HTTPException
from typing import Optional

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

# 定义依赖类：用户认证
class UserAuth:
    def __init__(self, authorization: Optional[str] = Header(None)):
        # 检查是否提供了 token
        if not authorization:
            raise HTTPException(
                status_code=401,
                detail="缺少认证信息"
            )

        # 提取 token
        if authorization.startswith("Bearer "):
            self.token = authorization[7:]
        else:
            self.token = authorization

        # 验证 token（简化示例）
        if self.token != "token123":
            raise HTTPException(
                status_code=401,
                detail="无效的认证 token"
            )

        # 设置用户信息
        self.user_id = 1
        self.username = "张三"

    def get_user_info(self):
        # 返回用户信息
        return {
            "user_id": self.user_id,
            "username": self.username
        }

# 定义路由：使用类依赖
# Depends(UserAuth) 会自动创建 UserAuth 实例
@app.get("/users/me")
async def read_users_me(auth: UserAuth = Depends(UserAuth)):
    # auth 是 UserAuth 的实例
    return {
        "message": "当前用户信息",
        "user": auth.get_user_info()
    }

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

11.6 嵌套依赖：依赖的依赖 #

依赖函数可以依赖其他依赖，形成依赖链。

# 导入 FastAPI 和 Depends
from fastapi import FastAPI, Depends, Header, HTTPException
from typing import Optional

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

# 模拟用户数据库
users_db = {
    "token123": {"id": 1, "name": "张三", "role": "admin"},
    "token456": {"id": 2, "name": "李四", "role": "user"}
}

# 第一层依赖：获取 token
async def get_token(authorization: Optional[str] = Header(None)):
    # 检查是否提供了 token
    if not authorization:
        raise HTTPException(status_code=401, detail="缺少认证信息")

    # 提取 token
    if authorization.startswith("Bearer "):
        return authorization[7:]
    return authorization

# 第二层依赖：获取当前用户（依赖于 get_token）
async def get_current_user(token: str = Depends(get_token)):
    # 验证 token
    if token not in users_db:
        raise HTTPException(status_code=401, detail="无效的 token")

    # 返回用户信息
    return users_db[token]

# 第三层依赖：检查管理员权限（依赖于 get_current_user）
async def require_admin(current_user: dict = Depends(get_current_user)):
    # 检查用户是否是管理员
    if current_user["role"] != "admin":
        raise HTTPException(
            status_code=403,
            detail="需要管理员权限"
        )
    return current_user

# 定义路由：普通用户可访问
@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(get_current_user)):
    # 只需要用户认证
    return {
        "message": "当前用户信息",
        "user": current_user
    }

# 定义路由：只有管理员可访问
@app.get("/admin/users")
async def get_all_users(admin: dict = Depends(require_admin)):
    # 需要管理员权限
    return {
        "message": "所有用户列表（管理员）",
        "users": list(users_db.values())
    }

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

说明：

• get_token：第一层依赖，获取 token
• get_current_user：第二层依赖，依赖于 get_token，验证用户
• require_admin：第三层依赖，依赖于 get_current_user，检查管理员权限

11.7 完整示例：数据库连接依赖 #

在实际项目中，数据库连接通常作为依赖注入。

# 导入 FastAPI 和 Depends
from fastapi import FastAPI, Depends, HTTPException
from typing import List, Optional

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

# 模拟数据库连接类
class Database:
    def __init__(self):
        # 模拟数据库连接
        self.items = [
            {"id": 1, "name": "苹果", "price": 5.5},
            {"id": 2, "name": "香蕉", "price": 3.0},
            {"id": 3, "name": "橙子", "price": 4.5}
        ]
        print("数据库连接已建立")

    def get_all_items(self):
        # 获取所有商品
        return self.items

    def get_item_by_id(self, item_id: int):
        # 根据 ID 获取商品
        for item in self.items:
            if item["id"] == item_id:
                return item
        return None

    def create_item(self, item: dict):
        # 创建新商品
        new_id = max([i["id"] for i in self.items]) + 1 if self.items else 1
        new_item = {"id": new_id, **item}
        self.items.append(new_item)
        return new_item

# 全局数据库实例（实际项目中应该使用连接池）
db_instance = None

# 定义依赖函数：获取数据库连接
async def get_db():
    global db_instance
    # 如果还没有创建数据库连接，创建一个
    if db_instance is None:
        db_instance = Database()
    # 返回数据库连接
    return db_instance

# 定义路由：获取所有商品
@app.get("/items/")
async def get_items(db: Database = Depends(get_db)):
    # db 是数据库连接实例
    items = db.get_all_items()
    return {
        "total": len(items),
        "items": items
    }

# 定义路由：根据 ID 获取商品
@app.get("/items/{item_id}")
async def get_item(item_id: int, db: Database = Depends(get_db)):
    # 使用数据库连接查询商品
    item = db.get_item_by_id(item_id)
    if not item:
        raise HTTPException(status_code=404, detail="商品不存在")
    return item

# 定义路由：创建新商品
@app.post("/items/")
async def create_item(
    name: str,
    price: float,
    db: Database = Depends(get_db)
):
    # 使用数据库连接创建商品
    new_item = db.create_item({"name": name, "price": price})
    return {
        "message": "商品创建成功",
        "item": new_item
    }

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

说明：

• get_db() 依赖函数负责创建和管理数据库连接
• 每个路由函数通过 Depends(get_db) 获取数据库连接
• 这样可以让代码更模块化，也便于测试（可以轻松替换为测试数据库）

11.8 常见使用场景 #

场景 1：用户认证 #

# 用户认证依赖
async def get_current_user(token: str = Header(...)):
    # 验证 token 并返回用户信息
    user = verify_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="无效的 token")
    return user

@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(get_current_user)):
    return current_user

场景 2：权限检查 #

# 权限检查依赖
async def require_admin(current_user: dict = Depends(get_current_user)):
    if current_user["role"] != "admin":
        raise HTTPException(status_code=403, detail="需要管理员权限")
    return current_user

@app.delete("/users/{user_id}")
async def delete_user(user_id: int, admin: dict = Depends(require_admin)):
    # 只有管理员可以删除用户
    return {"message": "用户已删除"}

场景 3：分页参数 #

# 分页参数依赖
async def get_pagination(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100)
):
    return {"skip": skip, "limit": limit}

@app.get("/items/")
async def get_items(pagination: dict = Depends(get_pagination)):
    skip = pagination["skip"]
    limit = pagination["limit"]
    # 使用分页参数查询数据
    return {"items": []}

11.9 Depends 的优势 #

使用 Depends 的好处：

1. 代码复用：将通用逻辑提取为依赖，多个路由可以共享
2. 代码清晰：路由函数只关注业务逻辑，依赖处理由 Depends 负责
3. 易于测试：可以轻松替换依赖为测试版本
4. 类型安全：FastAPI 会自动验证依赖的类型
5. 自动文档：依赖会自动出现在 API 文档中

12. 中间件：处理请求和响应 #

中间件（Middleware） 是 FastAPI 提供的强大功能，允许你在请求到达路由处理函数之前和响应返回给客户端之后执行代码。中间件可以用于日志记录、CORS 处理、请求计时、错误处理等全局性操作。

12.1 什么是中间件？ #

中间件是一个函数，它在每个请求处理之前和响应返回之后执行。中间件可以：

• 在请求到达路由之前处理请求（如添加请求头、记录日志）
• 在响应返回之前处理响应（如添加响应头、修改响应内容）
• 拦截请求并直接返回响应（如认证失败时）

简单理解：

• 路由函数：处理具体的业务逻辑（如获取用户信息）
• 中间件：处理所有请求的通用逻辑（如记录日志、添加 CORS 头）

中间件执行顺序：

请求 → 中间件1 → 中间件2 → ... → 路由函数 → ... → 中间件2 → 中间件1 → 响应

12.2 为什么需要中间件？ #

问题场景：假设你想记录每个请求的处理时间：

# 不好的做法：在每个路由函数中都要写计时逻辑
@app.get("/users/")
async def get_users():
    import time
    start_time = time.time()
    # 业务逻辑
    result = {"users": []}
    elapsed = time.time() - start_time
    print(f"处理时间：{elapsed:.2f}秒")
    return result

@app.get("/items/")
async def get_items():
    import time
    start_time = time.time()
    # 业务逻辑
    result = {"items": []}
    elapsed = time.time() - start_time
    print(f"处理时间：{elapsed:.2f}秒")
    return result

问题：

• 代码重复：每个路由都要写相同的计时逻辑
• 难以维护：如果要修改计时逻辑，需要修改所有路由
• 容易遗漏：新增路由时可能忘记添加计时逻辑

解决方案：使用中间件统一处理：

# 好的做法：使用中间件统一处理
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    import time
    start_time = time.time()
    response = await call_next(request)
    elapsed = time.time() - start_time
    response.headers["X-Process-Time"] = str(elapsed)
    return response

12.3 基本使用：创建中间件 #

中间件是一个异步函数，接收 Request 和 call_next 参数。

# 导入 FastAPI 和 Request
from fastapi import FastAPI, Request
from fastapi.responses import Response
import time

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

# 定义中间件：记录请求处理时间
# @app.middleware("http") 装饰器用于注册 HTTP 中间件
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    # request 是当前请求对象
    # call_next 是调用下一个中间件或路由函数的函数

    # 记录开始时间
    start_time = time.time()

    # 调用下一个中间件或路由函数，获取响应
    # await call_next(request) 会执行路由函数并返回响应
    response = await call_next(request)

    # 计算处理时间
    process_time = time.time() - start_time

    # 在响应头中添加处理时间
    # 这样客户端可以通过响应头知道服务器处理请求用了多长时间
    response.headers["X-Process-Time"] = str(process_time)

    # 返回响应
    return response

# 定义路由：获取用户列表
@app.get("/users/")
async def get_users():
    # 模拟一些处理时间
    await asyncio.sleep(0.1)
    return {"users": ["张三", "李四", "王五"]}

# 定义路由：获取商品列表
@app.get("/items/")
async def get_items():
    # 模拟一些处理时间
    await asyncio.sleep(0.2)
    return {"items": ["苹果", "香蕉", "橙子"]}

# 运行应用
if __name__ == "__main__":
    import uvicorn
    import asyncio
    uvicorn.run(app, host="0.0.0.0", port=8000)

说明：

• `@app.middleware("http")`：注册 HTTP 中间件
• request: Request：当前请求对象
• call_next：调用下一个中间件或路由函数的函数
• await call_next(request)：执行路由函数并获取响应
• 中间件可以在调用 call_next 之前和之后执行代码

测试方法：

# 访问路由，查看响应头中的 X-Process-Time
curl -i "http://localhost:8000/users/"
curl -i "http://localhost:8000/items/"

12.4 日志记录中间件 #

中间件常用于记录请求日志。

# 导入 FastAPI 和 Request
from fastapi import FastAPI, Request
import time
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

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

# 定义中间件：记录请求日志
@app.middleware("http")
async def log_requests(request: Request, call_next):
    # 记录请求开始
    start_time = time.time()

    # 记录请求信息
    logger.info(
        f"请求开始: {request.method} {request.url.path} "
        f"客户端: {request.client.host if request.client else 'unknown'}"
    )

    # 调用路由函数
    response = await call_next(request)

    # 计算处理时间
    process_time = time.time() - start_time

    # 记录请求完成信息
    logger.info(
        f"请求完成: {request.method} {request.url.path} "
        f"状态码: {response.status_code} "
        f"处理时间: {process_time:.3f}秒"
    )

    # 返回响应
    return response

# 定义路由
@app.get("/")
async def read_root():
    return {"message": "Hello World"}

@app.get("/users/")
async def get_users():
    return {"users": []}

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

运行效果：

INFO:     请求开始: GET / 客户端: 127.0.0.1
INFO:     请求完成: GET / 状态码: 200 处理时间: 0.001秒
INFO:     请求开始: GET /users/ 客户端: 127.0.0.1
INFO:     请求完成: GET /users/ 状态码: 200 处理时间: 0.002秒

12.5 CORS 中间件：跨域资源共享 #

CORS（Cross-Origin Resource Sharing）中间件用于处理跨域请求。

# 导入 FastAPI
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

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

# 添加 CORS 中间件
# CORSMiddleware 用于处理跨域请求
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 允许所有来源（生产环境应该指定具体域名）
    allow_credentials=True,  # 允许携带凭证（如 cookies）
    allow_methods=["*"],  # 允许所有 HTTP 方法
    allow_headers=["*"],  # 允许所有请求头
)

# 定义路由
@app.get("/")
async def read_root():
    return {"message": "Hello World"}

@app.get("/users/")
async def get_users():
    return {"users": ["张三", "李四"]}

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

说明：

• allow_origins：允许的源（域名列表），["*"] 表示允许所有域名
• allow_credentials：是否允许携带凭证（cookies、authorization headers）
• allow_methods：允许的 HTTP 方法，["*"] 表示允许所有方法
• allow_headers：允许的请求头，["*"] 表示允许所有请求头

生产环境建议：

# 生产环境应该指定具体的域名
app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "https://example.com",
        "https://www.example.com"
    ],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"],
)

12.6 自定义请求头中间件 #

中间件可以添加自定义请求头或修改响应头。

# 导入 FastAPI 和 Request
from fastapi import FastAPI, Request

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

# 定义中间件：添加自定义响应头
@app.middleware("http")
async def add_custom_headers(request: Request, call_next):
    # 调用路由函数获取响应
    response = await call_next(request)

    # 添加自定义响应头
    response.headers["X-Custom-Header"] = "Custom Value"
    response.headers["X-Server"] = "FastAPI"
    response.headers["X-Version"] = "1.0.0"

    # 返回响应
    return response

# 定义路由
@app.get("/")
async def read_root():
    return {"message": "Hello World"}

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

测试方法：

# 查看响应头
curl -i "http://localhost:8000/"

12.7 请求验证中间件 #

中间件可以用于验证请求（如检查 API 密钥）。

# 导入 FastAPI、Request 和 HTTPException
from fastapi import FastAPI, Request, HTTPException, Header
from fastapi.responses import JSONResponse

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

# 定义中间件：验证 API 密钥
@app.middleware("http")
async def verify_api_key(request: Request, call_next):
    # 跳过根路径和文档路径（不需要 API 密钥）
    if request.url.path in ["/", "/docs", "/openapi.json", "/redoc"]:
        # 直接调用下一个中间件或路由函数
        return await call_next(request)

    # 从请求头获取 API 密钥
    api_key = request.headers.get("X-API-Key")

    # 检查是否提供了 API 密钥
    if not api_key:
        # 返回 401 错误（未授权）
        return JSONResponse(
            status_code=401,
            content={"detail": "缺少 API 密钥，请提供 X-API-Key 头"}
        )

    # 验证 API 密钥（简化示例，实际应该从数据库或配置文件读取）
    valid_keys = ["secret-key-123", "secret-key-456"]
    if api_key not in valid_keys:
        # 返回 403 错误（禁止访问）
        return JSONResponse(
            status_code=403,
            content={"detail": "无效的 API 密钥"}
        )

    # API 密钥验证通过，继续处理请求
    response = await call_next(request)
    return response

# 定义路由：需要 API 密钥
@app.get("/users/")
async def get_users():
    return {"users": ["张三", "李四"]}

@app.get("/items/")
async def get_items():
    return {"items": ["苹果", "香蕉"]}

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

测试方法：

# 不带 API 密钥（会返回 401 错误）
curl "http://localhost:8000/users/"

# 带无效的 API 密钥（会返回 403 错误）
curl "http://localhost:8000/users/" -H "X-API-Key: invalid-key"

# 带有效的 API 密钥（成功）
curl "http://localhost:8000/users/" -H "X-API-Key: secret-key-123"

12.8 多个中间件：执行顺序 #

可以注册多个中间件，它们会按照注册的顺序执行。

# 导入 FastAPI 和 Request
from fastapi import FastAPI, Request
import time
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

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

# 第一个中间件：记录请求开始
@app.middleware("http")
async def middleware1(request: Request, call_next):
    logger.info("中间件1：请求开始")
    response = await call_next(request)
    logger.info("中间件1：请求完成")
    return response

# 第二个中间件：添加处理时间
@app.middleware("http")
async def middleware2(request: Request, call_next):
    start_time = time.time()
    logger.info("中间件2：开始计时")
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    logger.info(f"中间件2：处理时间 {process_time:.3f}秒")
    return response

# 第三个中间件：添加自定义头
@app.middleware("http")
async def middleware3(request: Request, call_next):
    logger.info("中间件3：添加响应头")
    response = await call_next(request)
    response.headers["X-Custom"] = "Value"
    logger.info("中间件3：响应头已添加")
    return response

# 定义路由
@app.get("/")
async def read_root():
    logger.info("路由函数：处理请求")
    return {"message": "Hello World"}

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

执行顺序：

请求 → 中间件1 → 中间件2 → 中间件3 → 路由函数 → 中间件3 → 中间件2 → 中间件1 → 响应

日志输出：

INFO:     中间件1：请求开始
INFO:     中间件2：开始计时
INFO:     中间件3：添加响应头
INFO:     路由函数：处理请求
INFO:     中间件3：响应头已添加
INFO:     中间件2：处理时间 0.001秒
INFO:     中间件1：请求完成

12.9 中间件与 Depends 的区别 #

中间件 vs Depends：

特性	中间件	Depends
执行时机	所有请求（包括静态文件、文档等）	特定路由函数
作用范围	全局	局部（特定路由）
用途	日志、CORS、请求验证等全局逻辑	认证、数据库连接等路由级逻辑
性能	每个请求都执行	只在需要时执行
灵活性	较低（全局应用）	较高（可以针对特定路由）

使用建议：

• 使用中间件：处理所有请求的全局逻辑（如日志、CORS、请求计时）
• 使用 Depends：处理特定路由的逻辑（如用户认证、数据库连接）

12.10 完整示例：综合使用中间件 #

这是一个完整的示例，展示如何综合使用多个中间件。

# 导入 FastAPI 和必要的模块
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
import time
import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# 创建应用实例
app = FastAPI(title="中间件示例")

# 添加 CORS 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 开发环境允许所有来源
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 定义中间件：记录请求日志
@app.middleware("http")
async def log_requests(request: Request, call_next):
    # 记录请求开始
    start_time = time.time()

    # 记录请求信息
    logger.info(
        f"{request.method} {request.url.path} - "
        f"客户端: {request.client.host if request.client else 'unknown'}"
    )

    # 调用路由函数
    response = await call_next(request)

    # 计算处理时间
    process_time = time.time() - start_time

    # 在响应头中添加处理时间
    response.headers["X-Process-Time"] = f"{process_time:.3f}"

    # 记录请求完成
    logger.info(
        f"{request.method} {request.url.path} - "
        f"状态码: {response.status_code} - "
        f"处理时间: {process_time:.3f}秒"
    )

    return response

# 定义中间件：添加自定义响应头
@app.middleware("http")
async def add_custom_headers(request: Request, call_next):
    # 调用路由函数
    response = await call_next(request)

    # 添加自定义响应头
    response.headers["X-Server"] = "FastAPI"
    response.headers["X-Version"] = "1.0.0"

    return response

# 定义路由
@app.get("/")
async def read_root():
    return {
        "message": "Hello World",
        "docs": "/docs"
    }

@app.get("/users/")
async def get_users():
    # 模拟一些处理时间
    await asyncio.sleep(0.1)
    return {"users": ["张三", "李四", "王五"]}

@app.get("/items/")
async def get_items():
    # 模拟一些处理时间
    await asyncio.sleep(0.2)
    return {"items": ["苹果", "香蕉", "橙子"]}

# 运行应用
if __name__ == "__main__":
    import uvicorn
    import asyncio
    uvicorn.run(app, host="0.0.0.0", port=8000)

运行效果：

• 所有请求都会记录日志
• 所有响应都会包含处理时间和自定义响应头
• 支持跨域请求（CORS）

13. Lifespan：应用生命周期管理 #

Lifespan（生命周期） 是 FastAPI 提供的功能，允许你在应用启动时和关闭时执行代码。这对于初始化资源（如数据库连接、缓存连接）和清理资源（如关闭连接、保存数据）非常有用。

13.1 什么是 Lifespan？ #

Lifespan 是一个异步上下文管理器，它定义了应用启动和关闭时执行的代码：

• 启动时（startup）：应用启动前执行，用于初始化资源
• 关闭时（shutdown）：应用关闭时执行，用于清理资源

简单理解：

• 中间件：处理每个请求的通用逻辑
• Lifespan：处理应用启动和关闭时的逻辑（只执行一次）

执行时机：

应用启动 → lifespan startup → 应用就绪 → 处理请求 → 应用关闭 → lifespan shutdown

13.2 为什么需要 Lifespan？ #

问题场景：假设你的应用需要连接数据库：

# 不好的做法：在路由函数中连接数据库
@app.get("/users/")
async def get_users():
    # 每次请求都要连接数据库（效率低）
    db = connect_to_database()
    users = db.query("SELECT * FROM users")
    db.close()  # 每次请求都要关闭连接
    return users

问题：

• 效率低：每次请求都要建立和关闭数据库连接
• 资源浪费：频繁创建和销毁连接
• 难以管理：连接状态分散在各个路由函数中

解决方案：使用 Lifespan 在应用启动时建立连接，关闭时清理：

# 好的做法：使用 Lifespan 管理数据库连接
@app.on_event("startup")
async def startup():
    app.state.db = connect_to_database()

@app.on_event("shutdown")
async def shutdown():
    app.state.db.close()

@app.get("/users/")
async def get_users():
    # 直接使用已建立的连接
    users = app.state.db.query("SELECT * FROM users")
    return users

13.3 基本使用：使用 @app.on_event 装饰器 #

FastAPI 提供了 `@app.on_event` 装饰器来定义启动和关闭事件。

# 导入 FastAPI
from fastapi import FastAPI

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

# 定义启动事件：应用启动时执行
@app.on_event("startup")
async def startup_event():
    # 在应用启动时执行的代码
    print("应用正在启动...")
    print("初始化数据库连接...")
    print("加载配置文件...")
    print("应用启动完成！")

# 定义关闭事件：应用关闭时执行
@app.on_event("shutdown")
async def shutdown_event():
    # 在应用关闭时执行的代码
    print("应用正在关闭...")
    print("关闭数据库连接...")
    print("保存缓存数据...")
    print("应用已关闭！")

# 定义路由
@app.get("/")
async def read_root():
    return {"message": "Hello World"}

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

运行效果：

• 启动应用时，会看到启动事件的输出
• 关闭应用时（按 Ctrl+C），会看到关闭事件的输出

13.4 使用 Lifespan 上下文管理器（推荐方式） #

FastAPI 推荐使用 lifespan 参数，它是一个异步上下文管理器，更清晰、更易管理。

# 导入 FastAPI 和必要的模块
from fastapi import FastAPI
from contextlib import asynccontextmanager

# 定义生命周期函数
# @asynccontextmanager 装饰器将函数转换为异步上下文管理器
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时执行的代码（yield 之前的代码）
    print("应用正在启动...")
    print("初始化数据库连接...")

    # 模拟数据库连接
    app.state.db_connected = True
    app.state.cache_connected = True

    print("应用启动完成！")

    # yield 表示应用已经就绪，可以处理请求了
    yield

    # 关闭时执行的代码（yield 之后的代码）
    print("应用正在关闭...")
    print("关闭数据库连接...")
    print("关闭缓存连接...")

    # 清理资源
    app.state.db_connected = False
    app.state.cache_connected = False

    print("应用已关闭！")

# 创建应用实例，传入 lifespan 参数
app = FastAPI(lifespan=lifespan)

# 定义路由
@app.get("/")
async def read_root():
    # 可以访问应用状态
    db_status = getattr(app.state, "db_connected", False)
    return {
        "message": "Hello World",
        "db_connected": db_status
    }

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

说明：

• @asynccontextmanager：将函数转换为异步上下文管理器
• yield：分隔启动代码和关闭代码
• yield 之前：应用启动时执行
• yield 之后：应用关闭时执行
• app.state：用于存储应用级别的状态（如数据库连接）

13.5 数据库连接示例 #

在实际项目中，lifespan 常用于管理数据库连接。

# 导入 FastAPI 和必要的模块
from fastapi import FastAPI
from contextlib import asynccontextmanager

# 模拟数据库连接类
class Database:
    def __init__(self):
        self.connected = False
        self.items = []

    async def connect(self):
        # 模拟连接数据库
        print("正在连接数据库...")
        await asyncio.sleep(0.1)  # 模拟连接耗时
        self.connected = True
        print("数据库连接成功！")
        # 初始化一些数据
        self.items = [
            {"id": 1, "name": "苹果", "price": 5.5},
            {"id": 2, "name": "香蕉", "price": 3.0}
        ]

    async def disconnect(self):
        # 模拟关闭数据库连接
        print("正在关闭数据库连接...")
        await asyncio.sleep(0.1)  # 模拟关闭耗时
        self.connected = False
        print("数据库连接已关闭！")

    def get_all_items(self):
        # 获取所有商品
        return self.items

    def get_item_by_id(self, item_id: int):
        # 根据 ID 获取商品
        for item in self.items:
            if item["id"] == item_id:
                return item
        return None

# 定义生命周期函数
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时：创建并连接数据库
    print("应用启动：初始化数据库...")
    db = Database()
    await db.connect()

    # 将数据库实例存储到 app.state 中
    app.state.db = db

    print("应用启动完成！")

    # yield 表示应用已就绪
    yield

    # 关闭时：关闭数据库连接
    print("应用关闭：清理数据库连接...")
    await app.state.db.disconnect()
    print("应用关闭完成！")

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

# 定义路由：获取所有商品
@app.get("/items/")
async def get_items():
    # 从 app.state 获取数据库实例
    db = app.state.db
    items = db.get_all_items()
    return {
        "total": len(items),
        "items": items
    }

# 定义路由：根据 ID 获取商品
@app.get("/items/{item_id}")
async def get_item(item_id: int):
    # 从 app.state 获取数据库实例
    db = app.state.db
    item = db.get_item_by_id(item_id)
    if not item:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail="商品不存在")
    return item

# 运行应用
if __name__ == "__main__":
    import uvicorn
    import asyncio
    uvicorn.run(app, host="0.0.0.0", port=8000)

说明：

• 数据库连接在应用启动时建立，所有路由共享同一个连接
• 数据库连接在应用关闭时自动清理
• 使用 app.state 存储应用级别的状态

13.6 多个启动和关闭任务 #

可以在 lifespan 中执行多个启动和关闭任务。

# 导入 FastAPI 和必要的模块
from fastapi import FastAPI
from contextlib import asynccontextmanager
import asyncio

# 模拟初始化任务
async def init_database():
    print("初始化数据库...")
    await asyncio.sleep(0.1)
    print("数据库初始化完成！")
    return {"db": "connected"}

async def init_cache():
    print("初始化缓存...")
    await asyncio.sleep(0.1)
    print("缓存初始化完成！")
    return {"cache": "connected"}

async def init_logging():
    print("初始化日志系统...")
    await asyncio.sleep(0.05)
    print("日志系统初始化完成！")
    return {"logging": "ready"}

# 模拟清理任务
async def cleanup_database(db_state):
    print("清理数据库连接...")
    await asyncio.sleep(0.1)
    print("数据库连接已清理！")

async def cleanup_cache(cache_state):
    print("清理缓存连接...")
    await asyncio.sleep(0.1)
    print("缓存连接已清理！")

# 定义生命周期函数
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时：并行执行多个初始化任务
    print("应用启动：开始初始化...")

    # 并行执行初始化任务（提高启动速度）
    db_state, cache_state, logging_state = await asyncio.gather(
        init_database(),
        init_cache(),
        init_logging()
    )

    # 将状态存储到 app.state
    app.state.db = db_state
    app.state.cache = cache_state
    app.state.logging = logging_state

    print("应用启动完成！所有服务已就绪。")

    # yield 表示应用已就绪
    yield

    # 关闭时：执行清理任务
    print("应用关闭：开始清理...")

    # 并行执行清理任务
    await asyncio.gather(
        cleanup_database(app.state.db),
        cleanup_cache(app.state.cache)
    )

    print("应用关闭完成！所有资源已清理。")

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

# 定义路由
@app.get("/")
async def read_root():
    return {
        "message": "Hello World",
        "status": {
            "db": app.state.db,
            "cache": app.state.cache,
            "logging": app.state.logging
        }
    }

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

说明：

• 使用 asyncio.gather() 并行执行多个初始化任务，提高启动速度
• 启动和关闭任务可以独立管理
• 所有状态都存储在 app.state 中

13.7 错误处理 #

在 lifespan 中，如果启动时发生错误，应用不会启动；如果关闭时发生错误，会记录日志但不会阻止关闭。

# 导入 FastAPI 和必要的模块
from fastapi import FastAPI
from contextlib import asynccontextmanager

# 定义生命周期函数（带错误处理）
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时：带错误处理
    try:
        print("应用启动：初始化服务...")

        # 模拟可能失败的操作
        # 如果这里抛出异常，应用不会启动
        app.state.service_ready = True
        print("服务初始化成功！")

    except Exception as e:
        # 如果启动失败，记录错误并重新抛出
        print(f"启动失败：{e}")
        raise

    # yield 表示应用已就绪
    yield

    # 关闭时：带错误处理
    try:
        print("应用关闭：清理资源...")
        # 清理操作
        app.state.service_ready = False
        print("资源清理成功！")

    except Exception as e:
        # 如果清理失败，记录错误但不阻止关闭
        print(f"清理时发生错误：{e}")

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

# 定义路由
@app.get("/")
async def read_root():
    service_status = getattr(app.state, "service_ready", False)
    return {
        "message": "Hello World",
        "service_ready": service_status
    }

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

13.8 Lifespan vs @app.on_event #

FastAPI 支持两种方式管理生命周期：

特性	Lifespan	@app.on_event
推荐程度	推荐	⚠️ 已弃用（FastAPI 0.95+）
代码组织	更清晰（启动和关闭代码在一起）	启动和关闭代码分离
错误处理	更容易处理	需要分别处理
类型提示	更好的类型支持	类型支持较弱

推荐使用 Lifespan：

• 代码更清晰：启动和关闭逻辑在一起
• 更好的错误处理：可以统一处理启动和关闭错误
• 符合现代 Python 最佳实践

13.9 完整示例：综合使用 Lifespan #

这是一个完整的示例，展示如何在实际项目中使用 Lifespan。

# 导入 FastAPI 和必要的模块
from fastapi import FastAPI, HTTPException
from contextlib import asynccontextmanager
import asyncio
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 模拟数据库连接类
class Database:
    def __init__(self):
        self.connected = False
        self.items = []
        self.next_id = 1

    async def connect(self):
        logger.info("正在连接数据库...")
        await asyncio.sleep(0.1)
        self.connected = True
        logger.info("数据库连接成功！")
        # 初始化示例数据
        self.items = [
            {"id": 1, "name": "苹果", "price": 5.5},
            {"id": 2, "name": "香蕉", "price": 3.0}
        ]
        self.next_id = 3

    async def disconnect(self):
        logger.info("正在关闭数据库连接...")
        await asyncio.sleep(0.1)
        self.connected = False
        logger.info("数据库连接已关闭！")

    def get_all_items(self):
        return self.items

    def get_item_by_id(self, item_id: int):
        for item in self.items:
            if item["id"] == item_id:
                return item
        return None

    def create_item(self, name: str, price: float):
        new_item = {
            "id": self.next_id,
            "name": name,
            "price": price
        }
        self.items.append(new_item)
        self.next_id += 1
        return new_item

# 模拟缓存连接类
class Cache:
    def __init__(self):
        self.connected = False
        self.data = {}

    async def connect(self):
        logger.info("正在连接缓存...")
        await asyncio.sleep(0.05)
        self.connected = True
        logger.info("缓存连接成功！")

    async def disconnect(self):
        logger.info("正在关闭缓存连接...")
        await asyncio.sleep(0.05)
        self.connected = False
        logger.info("缓存连接已关闭！")

    def get(self, key: str):
        return self.data.get(key)

    def set(self, key: str, value):
        self.data[key] = value

# 定义生命周期函数
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时：初始化数据库和缓存
    logger.info("=" * 50)
    logger.info("应用启动：开始初始化服务...")

    try:
        # 初始化数据库
        db = Database()
        await db.connect()
        app.state.db = db

        # 初始化缓存
        cache = Cache()
        await cache.connect()
        app.state.cache = cache

        logger.info("应用启动完成！所有服务已就绪。")
        logger.info("=" * 50)

    except Exception as e:
        logger.error(f"应用启动失败：{e}")
        raise

    # yield 表示应用已就绪，可以处理请求了
    yield

    # 关闭时：清理资源
    logger.info("=" * 50)
    logger.info("应用关闭：开始清理资源...")

    try:
        # 关闭缓存
        if hasattr(app.state, "cache"):
            await app.state.cache.disconnect()

        # 关闭数据库
        if hasattr(app.state, "db"):
            await app.state.db.disconnect()

        logger.info("应用关闭完成！所有资源已清理。")
        logger.info("=" * 50)

    except Exception as e:
        logger.error(f"清理资源时发生错误：{e}")

# 创建应用实例
app = FastAPI(
    title="Lifespan 示例",
    description="展示如何使用 Lifespan 管理应用生命周期",
    lifespan=lifespan
)

# 定义路由：获取所有商品
@app.get("/items/")
async def get_items():
    db = app.state.db
    items = db.get_all_items()
    return {
        "total": len(items),
        "items": items
    }

# 定义路由：根据 ID 获取商品
@app.get("/items/{item_id}")
async def get_item(item_id: int):
    db = app.state.db
    item = db.get_item_by_id(item_id)
    if not item:
        raise HTTPException(status_code=404, detail="商品不存在")
    return item

# 定义路由：创建新商品
@app.post("/items/")
async def create_item(name: str, price: float):
    db = app.state.db
    new_item = db.create_item(name, price)
    return {
        "message": "商品创建成功",
        "item": new_item
    }

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

运行效果：

• 启动应用时，会看到数据库和缓存的初始化日志
• 应用关闭时，会看到资源清理的日志
• 所有路由都可以使用 app.state.db 和 app.state.cache

13.10 常见使用场景 #

场景 1：数据库连接池 #

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时：创建数据库连接池
    app.state.db_pool = await create_db_pool()
    yield
    # 关闭时：关闭连接池
    await app.state.db_pool.close()

场景 2：加载配置文件 #

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时：加载配置文件
    app.state.config = load_config()
    yield
    # 关闭时：保存配置（如果需要）
    save_config(app.state.config)

场景 3：初始化外部服务 #

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时：连接外部服务（如 Redis、消息队列等）
    app.state.redis = await connect_redis()
    app.state.mq = await connect_message_queue()
    yield
    # 关闭时：断开连接
    await app.state.redis.close()
    await app.state.mq.close()

13.11 Lifespan 的优势 #

使用 Lifespan 的好处：

1. 资源管理：统一管理应用级别的资源（如数据库连接、缓存连接）
2. 性能优化：避免每次请求都创建和销毁资源
3. 代码清晰：启动和关闭逻辑集中管理
4. 错误处理：可以统一处理启动和关闭错误
5. 易于测试：可以轻松替换为测试版本

14. 错误处理：HTTPException #

当发生错误时，FastAPI 提供了 HTTPException 来返回标准的 HTTP 错误响应。

# 导入 FastAPI 和 HTTPException
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

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

# 定义用户模型
class User(BaseModel):
    username: str
    email: str

# 模拟用户数据库
users_db = {
    1: User(username="张三", email="zhangsan@example.com"),
    2: User(username="李四", email="lisi@example.com")
}

# 定义路由：获取用户信息
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # 检查用户是否存在
    if user_id not in users_db:
        # 抛出 404 错误（资源不存在）
        raise HTTPException(
            status_code=404,
            detail=f"用户 ID {user_id} 不存在"
        )
    # 返回用户信息
    return users_db[user_id]

# 定义路由：创建用户（带验证）
@app.post("/users/")
async def create_user(user: User):
    # 简单验证：检查用户名是否已存在
    for existing_user in users_db.values():
        if existing_user.username == user.username:
            # 抛出 400 错误（请求错误）
            raise HTTPException(
                status_code=400,
                detail="用户名已存在"
            )
    # 创建新用户
    new_id = max(users_db.keys()) + 1 if users_db else 1
    users_db[new_id] = user
    return {"message": "用户创建成功", "user_id": new_id, "user": user}

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

常见 HTTP 状态码：

• 200：成功
• 201：创建成功
• 400：请求错误（如参数验证失败）
• 404：资源不存在
• 500：服务器内部错误

15. 数据验证：使用 Pydantic 进行高级验证 #

Pydantic 提供了强大的数据验证功能，可以确保数据的正确性。

# 导入 FastAPI、Pydantic 和 Optional
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, EmailStr
from typing import Optional

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

# 定义用户模型，使用 Field 进行详细验证
class User(BaseModel):
    # name 字段：最小长度 2，最大长度 50
    name: str = Field(..., min_length=2, max_length=50, description="用户姓名")
    # email 字段：必须是有效的邮箱格式
    email: EmailStr = Field(..., description="用户邮箱")
    # age 字段：必须大于 0，小于 150
    age: int = Field(..., gt=0, lt=150, description="用户年龄")
    # phone 字段：可选，如果提供则必须是 11 位数字
    phone: Optional[str] = Field(None, min_length=11, max_length=11, description="手机号")

# 定义路由：创建用户
@app.post("/users/")
async def create_user(user: User):
    # FastAPI 会自动验证数据
    # 如果验证失败，会自动返回 422 错误（验证错误）
    return {
        "message": "用户创建成功",
        "user": user
    }

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Field 常用参数：

• ...：表示必填字段
• min_length：最小长度（字符串）
• max_length：最大长度（字符串）
• gt：大于（数字）
• lt：小于（数字）
• ge：大于等于（数字）
• le：小于等于（数字）
• description：字段描述（会显示在 API 文档中）

16. 自动生成 API 文档 #

FastAPI 的一个强大特性是自动生成交互式 API 文档，无需手动编写。

16.1 访问文档 #

启动应用后，访问以下 URL：

• Swagger UI：http://localhost:8000/docs

  • 交互式文档，可以直接测试 API

• ReDoc：http://localhost:8000/redoc

  • 更美观的文档界面，适合阅读

16.2 完整示例（带文档说明） #

# 导入 FastAPI、Pydantic 和 Optional
from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional

# 创建应用实例，可以添加标题和描述
app = FastAPI(
    title="我的 API",
    description="这是一个 FastAPI 教程示例",
    version="1.0.0"
)

# 定义商品模型，使用 Field 添加字段描述
class Item(BaseModel):
    name: str = Field(..., description="商品名称", example="苹果")
    price: float = Field(..., gt=0, description="商品价格（必须大于 0）", example=5.5)
    description: Optional[str] = Field(None, description="商品描述", example="新鲜苹果")

# 定义路由，添加 summary 和 description
@app.post(
    "/items/",
    summary="创建商品",
    description="创建一个新商品，需要提供商品名称和价格",
    response_description="返回创建的商品信息"
)
async def create_item(item: Item):
    return {
        "message": "商品创建成功",
        "item": item
    }

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

访问 http://localhost:8000/docs 可以看到完整的 API 文档，包括：

• 所有路由的列表
• 每个路由的参数说明
• 可以直接测试 API
• 请求和响应的示例

17. 常见错误与最佳实践 #

17.1 常见错误 #

错误 1：忘记导入必要的模块 #

# 错误示例
from fastapi import FastAPI

app = FastAPI()

@app.post("/items/")
async def create_item(item: Item):  # Item 未定义
    return item

# 正确示例
from fastapi import FastAPI
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float

app = FastAPI()

@app.post("/items/")
async def create_item(item: Item):
    return item

错误 2：路径参数和查询参数混淆 #

# 错误示例：路径参数应该放在 URL 路径中
@app.get("/items/{item_id}")
async def get_item(item_id: int, item_id: int):  # 重复定义
    return {"item_id": item_id}

# 正确示例
@app.get("/items/{item_id}")
async def get_item(item_id: int):  # 路径参数
    return {"item_id": item_id}

@app.get("/items/")
async def get_items(item_id: int):  # 查询参数
    return {"item_id": item_id}

错误 3：类型提示错误 #

# 错误示例：类型不匹配
@app.get("/items/{item_id}")
async def get_item(item_id: str):  # 路径参数是数字，但类型是字符串
    return {"item_id": item_id}

# 正确示例
@app.get("/items/{item_id}")
async def get_item(item_id: int):  # 类型匹配
    return {"item_id": item_id}

17.2 最佳实践 #

1. 使用类型提示：让代码更清晰，FastAPI 可以自动验证
2. 使用 Pydantic 模型：定义清晰的数据结构，自动验证数据
3. 添加文档说明：使用 summary、description 等参数，让 API 文档更清晰
4. 错误处理：使用 HTTPException 返回标准的错误响应
5. 代码组织：将模型、路由等分离到不同文件（项目较大时）

18. 总结 #

18.1 核心概念回顾 #

• FastAPI：现代、快速的 Python Web 框架
• 路由：使用装饰器（如 `@app.get()`）定义 API 端点
• 路径参数：URL 路径中的变量（如 /users/{user_id}）
• 查询参数：URL 中 ? 后面的参数（如 ?skip=0&limit=10）
• 请求体：POST/PUT 请求中发送的数据（JSON 格式）
• Pydantic 模型：用于定义和验证数据结构
• APIRouter：用于组织和管理路由，将不同功能的路由分组到不同模块
• Depends：依赖注入系统，用于处理依赖关系（如认证、数据库连接等）
• 中间件：用于处理所有请求的全局逻辑（如日志、CORS、请求计时）
• Lifespan：用于管理应用生命周期（启动时初始化资源，关闭时清理资源）
• HTTPException：用于返回 HTTP 错误响应

18.2 基本使用流程 #

1. 导入 FastAPI 和必要的模块
2. 创建 FastAPI 应用实例
3. 定义 Pydantic 模型（如果需要）
4. 使用装饰器定义路由（或使用 APIRouter 组织路由）
5. 在路由函数中处理请求并返回响应
6. 运行应用（使用 uvicorn）