FastAPI快速入门自我总结

FastAPI 精通教程

从入门到精通，自我总结，可能有缺，持续更新。。。

目录

  1. FastAPI 简介
  2. 快速开始
  3. 路径参数
  4. 查询参数
  5. 请求体与 Pydantic
  6. 响应模型
  7. 错误处理
  8. 依赖注入系统
  9. 中间件
  10. WebSocket 支持
  11. 后台任务
  12. 数据库集成
  13. 认证与授权
  14. 测试
  15. 项目结构最佳实践
  16. 性能优化

1. FastAPI 简介

什么是 FastAPI？

FastAPI 是一个现代、高性能的 Python Web 框架，用于构建 API。它的核心特点：

特性 说明
🚀 高性能 基于 Starlette 和 Pydantic，性能接近 Node.js 和 Go
📝 自动文档 自动生成 OpenAPI 和 ReDoc 文档
✅ 数据验证 基于 Python 类型提示的自动验证
🎯 开发速度快 代码量减少约 40%，bug 减少
🔧 生产就绪 支持异步、WebSocket、GraphQL 等

为什么选择 FastAPI？

1
2
3
4
5
传统框架 (Flask/Django)     FastAPI
├── 手动验证输入            ├── 自动类型验证
├── 手动写 API 文档          ├── 自动生成文档
├── 同步处理为主            ├── 原生异步支持
└── 返回值无类型检查        └── 响应模型验证

核心概念一图流

1
2
3
4
请求 → 路由 (Router) → 依赖注入 → 端点函数 → 响应模型 → JSON
              ↓              ↓           ↓
         路径参数       数据库连接    数据验证
         查询参数       认证检查     序列化

2. 快速开始

安装

1
pip install fastapi uvicorn[standard]

Hello World

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from fastapi import FastAPI

app = FastAPI(
    title="我的 API",
    description="这是一个示例 API",
    version="1.0.0"
)

@app.get("/")
async def root():
    """
    根路径端点
    
    这是你的第一个 FastAPI 端点！
    async def 表示这是一个异步函数
    """
    return {"message": "Hello World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    """
    带路径参数的端点
    
    item_id: 会自动转换为整数，如果传入了非整数会返回验证错误
    """
    return {"item_id": item_id}

运行服务器

1
2
3
4
5
6
7
8
# 开发模式（自动重载）
uvicorn main:app --reload

# 生产模式
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

# 指定端口
uvicorn main:app --port 8080

访问自动文档

  • Swagger UI: http://localhost:8000/docs
  • ReDoc: http://localhost:8000/redoc
  • OpenAPI JSON: http://localhost:8000/openapi.json

3. 路径参数

路径参数是 URL 路径的一部分，用于标识特定资源。

基础用法

1
2
3
4
5
6
7
8
9
10
11
12
13
from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    """
    获取用户信息
    
    user_id 会从 URL 中提取并验证为整数
    例如：/users/123 → user_id = 123
    """
    return {"user_id": user_id}

类型转换详解

FastAPI 会自动将路径参数转换为指定的类型：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from datetime import datetime
from uuid import UUID

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    # item_id 被转换为 int
    # /items/42 ✓ → {"item_id": 42}
    # /items/abc ✗ → 422 验证错误
    return {"item_id": item_id}

@app.get("/users/{user_id}")
async def read_user(user_id: UUID):
    # UUID 类型验证
    # /users/123e4567-e89b-12d3-a456-426614174000 ✓
    return {"user_id": user_id}

@app.get("/logs/{date}")
async def read_logs(date: datetime):
    # 日期时间验证
    # /logs/2024-01-15T10:30:00 ✓
    return {"date": date}

预定义值（Enum）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from enum import Enum
from fastapi import FastAPI

app = FastAPI()

class ModelName(str, Enum):
    """定义允许的值"""
    alexnet = "alexnet"
    resnet = "resnet"
    squeezenet = "squeezenet"

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    """
    model_name 只能是 ModelName 中定义的值
    
    /models/alexnet ✓
    /models/vgg ✗ → 422 错误
    """
    return {"model_name": model_name, "message": f"Using {model_name.value}"}

文件路径参数

1
2
3
4
5
6
7
8
9
10
11
12
13
from fastapi import FastAPI

app = FastAPI()

# 捕获剩余路径
@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    """
    file_path:path 会捕获剩余的所有路径
    
    /files/users/john/doc.txt → file_path = "users/john/doc.txt"
    """
    return {"file_path": file_path}

4. 查询参数

查询参数是 URL ? 后面的键值对。

基础用法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from fastapi import FastAPI
from typing import Optional

app = FastAPI()

@app.get("/items")
async def read_items(
    skip: int = 0,      # 默认值为 0
    limit: int = 10,    # 默认值为 10
    q: Optional[str] = None  # 可选参数
):
    """
    查询参数示例
    
    /items?skip=0&limit=10&q=search
    /items → skip=0, limit=10, q=None
    """
    return {"skip": skip, "limit": limit, "q": q}

必需参数 vs 可选参数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from fastapi import FastAPI

app = FastAPI()

# 必需参数（没有默认值）
@app.get("/users")
async def get_users(region: str):
    """
    /users?region=beijing ✓
    /users ✗ → 缺少必需参数
    """
    return {"region": region}

# 可选参数（有默认值）
@app.get("/products")
async def get_products(
    category: str = "all",  # 有默认值=可选
    page: int = 1,
    sort: str = "default"
):
    """
    /products ✓ → category="all", page=1, sort="default"
    /products?category=electronics&page=2 ✓
    """
    return {"category": category, "page": page, "sort": sort}

类型验证和约束

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from fastapi import FastAPI, Query
from typing import Optional, List

app = FastAPI()

@app.get("/items")
async def read_items(
    # 使用 Query 添加验证规则
    q: Optional[str] = Query(
        None,
        min_length=1,      # 最小长度
        max_length=50,     # 最大长度
        pattern=r"^[a-z]+$"  # 正则表达式
    ),
    # 数值范围
    price_min: float = Query(0, ge=0),  # ge: greater or equal
    price_max: float = Query(1000, le=10000),  # le: less or equal
    # 列表参数
    tags: List[str] = Query(default=[]),  # /items?tags=tag1&tags=tag2
    # 描述（显示在文档中）
    sort_by: str = Query("created_at", description="排序字段")
):
    """带验证的查询参数"""
    return {
        "q": q,
        "price_range": [price_min, price_max],
        "tags": tags,
        "sort_by": sort_by
    }

别名和废弃参数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items")
async def read_items(
    # 使用别名（URL 中用 page-num，函数中用 page_num）
    page_num: int = Query(1, alias="page-num"),
    # 标记为废弃
    old_param: str = Query(
        None,
        deprecated=True,
        description="此参数已废弃，请使用 new_param"
    ),
    new_param: str = Query(None)
):
    return {"page": page_num, "old": old_param, "new": new_param}

5. 请求体与 Pydantic

请求体用于发送复杂数据（通常是 JSON）。

Pydantic 模型基础

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
from fastapi import FastAPI
from pydantic import BaseModel, Field, EmailStr
from typing import Optional
from datetime import datetime

app = FastAPI()

# 定义数据模型
class Item(BaseModel):
    """
    Pydantic 模型 = 数据验证 + 序列化
    
    作用：
    1. 验证传入的数据类型
    2. 自动转换为正确的类型
    3. 生成 OpenAPI 文档
    """
    name: str = Field(..., min_length=1, max_length=50)
    price: float = Field(..., gt=0, description="价格必须大于 0")
    description: Optional[str] = None
    quantity: int = Field(1, ge=1)  # 默认值 1，必须 >= 1
    tags: list[str] = []

@app.post("/items")
async def create_item(item: Item):
    """
    FastAPI 会：
    1. 读取 JSON 请求体
    2. 验证数据类型
    3. 转换为 Item 对象
    4. 验证失败时返回 422
    """
    # item 已经是验证过的 Item 对象
    return {
        "name": item.name,
        "price": item.price,
        "total": item.price * item.quantity
    }

嵌套模型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from pydantic import BaseModel
from typing import Optional

class Address(BaseModel):
    street: str
    city: str
    country: str
    zipcode: str

class User(BaseModel):
    name: str
    email: str
    age: int
    address: Optional[Address] = None  # 嵌套模型

@app.post("/users")
async def create_user(user: User):
    """
    发送 JSON:
    {
        "name": "John",
        "email": "[email protected]",
        "age": 30,
        "address": {
            "street": "123 Main St",
            "city": "NYC",
            "country": "USA",
            "zipcode": "10001"
        }
    }
    """
    return user

验证器（Validators）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from pydantic import BaseModel, field_validator, model_validator
from datetime import datetime

class Product(BaseModel):
    name: str
    price: float
    discount: float = 0
    final_price: float = None

    # 字段验证器
    @field_validator('name')
    @classmethod
    def name_must_not_be_empty(cls, v):
        if not v.strip():
            raise ValueError('name 不能为空')
        return v.strip()

    @field_validator('price')
    @classmethod
    def price_must_be_positive(cls, v):
        if v <= 0:
            raise ValueError('价格必须大于 0')
        return v

    # 模型验证器（跨字段验证）
    @model_validator(mode='after')
    def calculate_final_price(self):
        if self.final_price is None:
            self.final_price = self.price * (1 - self.discount / 100)
        return self

@app.post("/products")
async def create_product(product: Product):
    return product

请求体 + 路径 + 查询参数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from fastapi import FastAPI, Path, Query
from pydantic import BaseModel

app = FastAPI()

class ItemUpdate(BaseModel):
    name: str | None = None
    price: float | None = None

@app.put("/items/{item_id}")
async def update_item(
    item_id: int = Path(..., description="要更新的物品 ID"),
    q: str | None = Query(None, description="查询参数"),
    item: ItemUpdate = None  # 请求体
):
    """
    PUT /items/123?q=test
    Body: {"name": "新名字", "price": 99.9}
    
    参数顺序：Path → Query → Body
    """
    return {"item_id": item_id, "q": q, "update": item}

6. 响应模型

响应模型用于控制返回给客户端的数据。

基础响应模型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
from typing import Optional

app = FastAPI()

class UserIn(BaseModel):
    """输入模型"""
    username: str
    password: str
    email: EmailStr

class UserOut(BaseModel):
    """输出模型 - 不包含密码"""
    username: str
    email: EmailStr
    
    class Config:
        from_attributes = True  # 允许从 ORM 对象读取

@app.post("/users", response_model=UserOut)
async def create_user(user: UserIn):
    """
    即使返回完整的 user 对象
    响应模型会过滤掉 password 字段
    """
    return user  # 实际返回只有 username 和 email

# 或者使用装饰器方式
@app.post("/users2")
async def create_user2(user: UserIn):
    return UserOut(**user.model_dump())

响应模型参数详解

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from fastapi import FastAPI
from pydantic import BaseModel, SecretStr
from typing import Optional, List

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: float = 10.0
    tags: List[str] = []

# exclude: 排除特定字段
# exclude_none: 排除值为 None 的字段
# exclude_unset: 只返回设置的字段

@app.post("/items", response_model=Item, response_model_exclude_none=True)
async def create_item(item: Item):
    """排除 None 值的字段"""
    return item

@app.get("/items/{item_id}", response_model=Item, response_model_exclude={"tax"})
async def get_item(item_id: int):
    """排除 tax 字段"""
    return Item(name=f"Item {item_id}", price=10.0)

返回状态码和自定义响应

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from fastapi import FastAPI, status, HTTPException

app = FastAPI()

@app.post("/items", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):
    """返回 201 Created 状态码"""
    return {"name": name}

@app.delete("/items/{item_id}", status_code=204)
async def delete_item(item_id: int):
    """返回 204 No Content"""
    return None

@app.post("/login")
async def login(username: str, password: str):
    if username != "admin" or password != "secret":
        raise HTTPException(
            status_code=401,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"}
        )
    return {"token": "fake-token"}

7. 错误处理

HTTPException

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from fastapi import FastAPI, HTTPException, status

app = FastAPI()

items = {"foo": "Item Foo", "bar": "Item Bar"}

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item {item_id} not found",
            headers={"X-Error": "NotFound"}  # 自定义头
        )
    return {"item": items[item_id]}

自定义异常处理器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

app = FastAPI()

class CustomException(Exception):
    def __init__(self, message: str, code: str):
        self.message = message
        self.code = code

# 注册异常处理器
@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=400,
        content={
            "error": "custom_error",
            "message": exc.message,
            "code": exc.code
        }
    )

# 重写默认验证错误处理
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={
            "error": "validation_error",
            "details": exc.errors()
        }
    )

@app.get("/trigger-error")
async def trigger_error():
    raise CustomException("这是一个自定义错误", "CUSTOM_001")

业务异常封装

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
from fastapi import FastAPI, HTTPException, status
from typing import Generic, TypeVar, Optional
from pydantic import BaseModel

app = FastAPI()

T = TypeVar('T')

# 统一响应格式
class APIResponse(BaseModel, Generic[T]):
    success: bool
    data: Optional[T] = None
    error: Optional[str] = None
    message: str = "操作成功"

class BusinessException(HTTPException):
    """业务异常基类"""
    def __init__(self, message: str, status_code: int = 400, code: str = "BIZ_ERROR"):
        super().__init__(
            status_code=status_code,
            detail={"message": message, "code": code}
        )

class NotFoundError(BusinessException):
    def __init__(self, resource: str):
        super().__init__(
            message=f"{resource} 未找到",
            status_code=404,
            code="NOT_FOUND"
        )

class ValidationError(BusinessException):
    def __init__(self, field: str, message: str):
        super().__init__(
            message=f"{field}: {message}",
            status_code=422,
            code="VALIDATION_ERROR"
        )

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    if user_id <= 0:
        raise ValidationError("user_id", "必须是正整数")
    # 模拟数据库查询
    if user_id == 999:
        raise NotFoundError("用户")
    return {"user_id": user_id}

8. 依赖注入系统

依赖注入是 FastAPI 最强大的特性之一。

基础依赖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from fastapi import FastAPI, Depends, HTTPException
from typing import Optional

app = FastAPI()

# 可重用的依赖
async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
    """返回通用参数"""
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items")
async def read_items(commons: dict = Depends(common_parameters)):
    """使用依赖注入"""
    return commons

@app.get("/users")
async def read_users(commons: dict = Depends(common_parameters)):
    """相同的依赖可以重复使用"""
    return {**commons, "type": "users"}

类依赖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
from fastapi import FastAPI, Depends

app = FastAPI()

class Pagination:
    """分页依赖"""
    def __init__(self, page: int = 1, page_size: int = 10):
        self.page = page
        self.page_size = page_size
    
    @property
    def offset(self) -> int:
        return (self.page - 1) * self.page_size

class Sorting:
    """排序依赖"""
    def __init__(self, sort_by: str = "created_at", order: str = "desc"):
        self.sort_by = sort_by
        self.order = order

@app.get("/items")
async def read_items(
    pagination: Pagination = Depends(),
    sorting: Sorting = Depends()
):
    """使用类依赖"""
    return {
        "page": pagination.page,
        "page_size": pagination.page_size,
        "offset": pagination.offset,
        "sort_by": sorting.sort_by,
        "order": sorting.order
    }

数据库会话依赖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
from fastapi import FastAPI, Depends
from typing import Generator

app = FastAPI()

# 模拟数据库连接
class DatabaseSession:
    def __init__(self, url: str = "sqlite:///:memory:"):
        self.url = url
        self.connected = False
    
    def connect(self):
        self.connected = True
        print(f"连接到数据库：{self.url}")
    
    def close(self):
        self.connected = False
        print("关闭数据库连接")

def get_db() -> Generator[DatabaseSession, None, None]:
    """
    依赖生成器 - 管理数据库会话生命周期
    
    yield 之前的代码：请求前执行（连接数据库）
    yield: 提供会话对象给端点使用
    yield 之后的代码：请求后执行（关闭连接）
    """
    db = DatabaseSession()
    try:
        db.connect()
        yield db
    finally:
        db.close()

@app.get("/users")
async def read_users(db: DatabaseSession = Depends(get_db)):
    """自动获取和清理数据库连接"""
    assert db.connected, "数据库应该已连接"
    return {"users": [], "db_connected": db.connected}

依赖嵌套和覆盖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
from fastapi import FastAPI, Depends, HTTPException, status

app = FastAPI()

# 用户认证依赖
async def get_current_user(token: str = None):
    if token is None:
        raise HTTPException(status_code=401, detail="缺少 Token")
    if token != "valid-token":
        raise HTTPException(status_code=401, detail="无效 Token")
    return {"username": "testuser", "role": "user"}

# 依赖上面的依赖
async def get_current_active_user(
    current_user: dict = Depends(get_current_user)
):
    if current_user.get("disabled"):
        raise HTTPException(status_code=400, detail="用户已禁用")
    return current_user

# 进一步依赖
async def get_current_admin_user(
    current_user: dict = Depends(get_current_active_user)
):
    if current_user.get("role") != "admin":
        raise HTTPException(status_code=403, detail="需要管理员权限")
    return current_user

@app.get("/me")
async def read_me(current_user: dict = Depends(get_current_active_user)):
    """需要活跃用户"""
    return current_user

@app.get("/admin/dashboard")
async def admin_dashboard(current_user: dict = Depends(get_current_admin_user)):
    """需要管理员"""
    return {"message": f"欢迎管理员 {current_user['username']}"}

依赖覆盖（测试用）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from fastapi import FastAPI, Depends

app = FastAPI()

def get_dependency():
    return "production_value"

@app.get("/")
async def read_root(value: str = Depends(get_dependency)):
    return {"value": value}

# 测试时覆盖依赖
# from .main import app, get_dependency
# async def override_dependency():
#     return "test_value"
# app.dependency_overrides[get_dependency] = override_dependency

9. 中间件

中间件在每个请求到达端点之前和响应返回之后执行。

基础中间件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time

app = FastAPI()

# 添加内置中间件
app.add_middleware(
    CORSMiddleware,  # 跨域支持
    allow_origins=["*"],  # 允许的来源
    allow_credentials=True,  # 允许 cookies
    allow_methods=["*"],  # 允许的方法
    allow_headers=["*"],  # 允许的头
)

# 自定义中间件
@app.middleware("http")
async def log_requests(request: Request, call_next):
    """记录请求日志的中间件"""
    start_time = time.time()
    
    # 请求前处理
    print(f"收到请求：{request.method} {request.url.path}")
    
    # 处理请求
    response = await call_next(request)
    
    # 响应后处理
    process_time = time.time() - start_time
    print(f"处理时间：{process_time:.3f}s")
    
    # 添加自定义响应头
    response.headers["X-Process-Time"] = str(process_time)
    
    return response

认证中间件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
from fastapi import FastAPI, Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse

class AuthenticationMiddleware(BaseHTTPMiddleware):
    """JWT 认证中间件"""
    
    async def dispatch(self, request: Request, call_next):
        # 跳过不需要认证的路径
        if request.url.path in ["/docs", "/openapi.json", "/login"]:
            return await call_next(request)
        
        # 检查 Authorization 头
        auth_header = request.headers.get("Authorization")
        if not auth_header or not auth_header.startswith("Bearer "):
            return JSONResponse(
                status_code=401,
                content={"detail": "缺少认证令牌"}
            )
        
        token = auth_header.split(" ")[1]
        
        # 验证 token（这里简化处理）
        if not self.verify_token(token):
            return JSONResponse(
                status_code=401,
                content={"detail": "无效的认证令牌"}
            )
        
        # 将用户信息添加到请求
        request.state.user = self.decode_token(token)
        
        return await call_next(request)
    
    def verify_token(self, token: str) -> bool:
        # 实际的 token 验证逻辑
        return token == "valid-token"
    
    def decode_token(self, token: str) -> dict:
        # 实际的 token 解码逻辑
        return {"sub": "user123", "exp": 9999999999}

app.add_middleware(AuthenticationMiddleware)

速率限制中间件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
from collections import defaultdict
import time

class RateLimitMiddleware(BaseHTTPMiddleware):
    """简单的速率限制"""
    
    def __init__(self, app, requests_per_minute: int = 60):
        super().__init__(app)
        self.requests_per_minute = requests_per_minute
        self.request_counts = defaultdict(list)
    
    async def dispatch(self, request: Request, call_next):
        client_ip = request.client.host
        current_time = time.time()
        window_start = current_time - 60
        
        # 清理旧请求
        self.request_counts[client_ip] = [
            t for t in self.request_counts[client_ip] if t > window_start
        ]
        
        # 检查是否超出限制
        if len(self.request_counts[client_ip]) >= self.requests_per_minute:
            return JSONResponse(
                status_code=429,
                content={"detail": "请求过于频繁，请稍后再试"}
            )
        
        # 记录请求
        self.request_counts[client_ip].append(current_time)
        
        return await call_next(request)

app.add_middleware(RateLimitMiddleware, requests_per_minute=100)

10. WebSocket 支持

基础 WebSocket

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    """WebSocket 端点"""
    await websocket.accept()
    
    while True:
        # 接收消息
        data = await websocket.receive_text()
        # 处理并返回
        response = f"收到：{data}"
        await websocket.send_text(response)

WebSocket + 依赖注入

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from fastapi import FastAPI, WebSocket, Depends, HTTPException
from typing import Optional

app = FastAPI()

async def get_token(websocket: WebSocket) -> str:
    """从 WebSocket 连接中获取 token"""
    token = websocket.query_params.get("token")
    if not token or token != "secret-token":
        await websocket.close(code=4001)
        raise HTTPException(status_code=400, detail="无效的 Token")
    return token

@app.websocket("/ws/auth")
async def authenticated_websocket(
    websocket: WebSocket,
    token: str = Depends(get_token)
):
    """需要认证的 WebSocket"""
    await websocket.accept()
    await websocket.send_text(f"欢迎！Token: {token}")
    
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Echo: {data}")

WebSocket 聊天室示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List
import json

app = FastAPI()

class ConnectionManager:
    """管理 WebSocket 连接"""
    
    def __init__(self):
        self.active_connections: List[WebSocket] = []
    
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
    
    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
    
    async def broadcast(self, message: str):
        """广播消息给所有连接"""
        for connection in self.active_connections:
            await connection.send_text(message)
    
    async def send_personal(self, message: str, websocket: WebSocket):
        """发送个人消息"""
        await websocket.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/chat")
async def chat_websocket(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        await manager.broadcast(f"新用户加入聊天室")
        while True:
            data = await websocket.receive_text()
            message = json.loads(data)
            await manager.broadcast(
                json.dumps({
                    "user": message.get("user", "匿名"),
                    "message": message.get("text", "")
                })
            )
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast("用户离开了聊天室")

11. 后台任务

基础后台任务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from fastapi import FastAPI, BackgroundTasks
import time

app = FastAPI()

def send_email(email: str, message: str):
    """模拟发送邮件"""
    print(f"发送邮件到 {email}: {message}")
    time.sleep(2)  # 模拟耗时操作
    print("邮件发送完成")

@app.post("/send-email")
async def send_email_endpoint(
    email: str,
    background_tasks: BackgroundTasks
):
    """
    后台任务会在响应返回后异步执行
    用户会立即收到响应，而不用等待邮件发送完成
    """
    background_tasks.add_task(send_email, email, "欢迎注册！")
    return {"message": "邮件正在发送中", "email": email}

@app.post("/multiple-tasks")
async def multiple_background_tasks(
    background_tasks: BackgroundTasks
):
    """添加多个后台任务"""
    background_tasks.add_task(send_email, "[email protected]", "通知 1")
    background_tasks.add_task(send_email, "[email protected]", "通知 2")
    background_tasks.add_task(send_email, "[email protected]", "通知 3")
    return {"message": "3 个邮件正在发送"}

使用 Celery 处理重型任务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
# 重型任务建议使用 Celery
# pip install celery[redis]

from celery import Celery
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

# Celery 配置
celery_app = Celery(
    'tasks',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/0'
)

@celery_app.task
def process_large_file(file_path: str):
    """处理大文件的 Celery 任务"""
    # 耗时操作
    time.sleep(10)
    return {"processed": file_path}

@app.post("/process-file")
async def process_file(file_path: str):
    """
    返回任务 ID，用户可以稍后查询结果
    """
    task = process_large_file.delay(file_path)
    return {"task_id": task.id, "status": "processing"}

@app.get("/task-status/{task_id}")
async def get_task_status(task_id: str):
    from celery.result import AsyncResult
    task_result = AsyncResult(task_id, app=celery_app)
    return {
        "task_id": task_id,
        "status": task_result.status,
        "result": task_result.result if task_result.ready() else None
    }

12. 数据库集成

SQLAlchemy 集成

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel
from typing import Optional, List

app = FastAPI()

# 数据库配置
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 模型定义
class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)

# 创建表
Base.metadata.create_all(bind=engine)

# Pydantic 模型
class UserBase(BaseModel):
    username: str
    email: str

class UserCreate(UserBase):
    password: str

class UserResponse(UserBase):
    id: int
    
    class Config:
        from_attributes = True

# 依赖项
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# CRUD 端点
@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(
        username=user.username,
        email=user.email,
        hashed_password=user.password  # 实际应使用 bcrypt 加密
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@app.get("/users", response_model=List[UserResponse])
async def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = db.query(User).offset(skip).limit(limit).all()
    return users

@app.get("/users/{user_id}", response_model=UserResponse)
async def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    return user

Tortoise ORM（异步）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
from fastapi import FastAPI
from tortoise import fields, models
from tortoise.contrib.fastapi import register_tortoise

app = FastAPI()

class User(models.Model):
    id = fields.IntField(pk=True)
    username = fields.CharField(max_length=50, unique=True)
    email = fields.CharField(max_length=100)
    created_at = fields.DatetimeField(auto_now_add=True)
    
    class Meta:
        table = "users"

@app.get("/users")
async def get_users():
    users = await User.all()
    return [{"id": u.id, "username": u.username} for u in users]

@app.post("/users")
async def create_user(username: str, email: str):
    user = await User.create(username=username, email=email)
    return {"id": user.id, "username": user.username}

# 注册 Tortoise
register_tortoise(
    app,
    db_url="sqlite://:memory:",
    modules={"models": ["__main__"]},
    generate_schemas=True,
    add_exception_handlers=True
)

13. 认证与授权

JWT 认证

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from typing import Optional
from datetime import datetime, timedelta
from jose import JWTError, jwt

app = FastAPI()

# 配置
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# 模拟用户数据库
fake_users_db = {
    "admin": {
        "username": "admin",
        "password": "secret",  # 实际应使用 bcrypt
        "disabled": False
    }
}

# OAuth2 scheme
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class Token(BaseModel):
    access_token: str
    token_type: str

class User(BaseModel):
    username: str
    disabled: Optional[bool] = None

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    
    user = fake_users_db.get(username)
    if user is None:
        raise credentials_exception
    return User(**user)

@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = fake_users_db.get(form_data.username)
    if not user or user["password"] != form_data.password:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    access_token = create_access_token(
        data={"sub": user["username"]},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return {"access_token": access_token, "token_type": "bearer"}

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

权限系统

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from typing import List

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class Permissions:
    ADMIN = "admin"
    USER_READ = "user:read"
    USER_WRITE = "user:write"

def require_permissions(required_permissions: List[str]):
    async def permission_checker(current_user: dict = Depends(get_current_user)):
        user_permissions = current_user.get("permissions", [])
        
        # 检查是否有所有必需的权限
        for perm in required_permissions:
            if perm not in user_permissions and Permissions.ADMIN not in user_permissions:
                raise HTTPException(
                    status_code=403,
                    detail=f"缺少权限：{perm}"
                )
        return current_user
    return permission_checker

@app.get("/admin/users")
async def list_all_users(
    current_user: dict = Depends(require_permissions([Permissions.USER_READ]))
):
    return {"message": "用户列表", "user": current_user["username"]}

@app.post("/admin/users")
async def create_user_endpoint(
    current_user: dict = Depends(require_permissions([Permissions.USER_WRITE]))
):
    return {"message": "创建用户", "user": current_user["username"]}

14. 测试

使用 TestClient

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from fastapi.testclient import TestClient
from fastapi import FastAPI

app = FastAPI()

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

@app.post("/items")
async def create_item(name: str, price: float):
    return {"name": name, "price": price}

# 测试
client = TestClient(app)

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

def test_create_item():
    response = client.post("/items?name=Apple&price=1.5")
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "Apple"
    assert data["price"] == 1.5

def test_validation_error():
    response = client.post("/items?name=Test")
    assert response.status_code == 422

测试带依赖的端点

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from fastapi import FastAPI, Depends
from fastapi.testclient import TestClient
from unittest.mock import Mock

app = FastAPI()

def get_db():
    return {"connection": "real_db"}

async def get_current_user(token: str):
    return {"username": "testuser"}

@app.get("/protected")
async def protected_route(user: dict = Depends(get_current_user)):
    return {"user": user}

# 覆盖依赖进行测试
def test_protected_route():
    app.dependency_overrides[get_current_user] = lambda: {"username": "test_user"}
    
    client = TestClient(app)
    response = client.get("/protected")
    
    assert response.status_code == 200
    assert response.json()["user"]["username"] == "test_user"
    
    # 清理覆盖
    app.dependency_overrides.clear()

Pytest 集成

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# test_main.py
import pytest
from fastapi.testclient import TestClient
from main import app

@pytest.fixture
def client():
    return TestClient(app)

def test_read_main(client):
    response = client.get("/")
    assert response.status_code == 200

def test_read_item(client):
    response = client.get("/items/1")
    assert response.status_code == 200

15. 项目结构最佳实践

推荐的项目结构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
my_project/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI 应用实例
│   ├── config.py            # 配置管理
│   ├── dependencies.py      # 依赖注入
│   │
│   ├── models/              # 数据库模型
│   │   ├── __init__.py
│   │   └── user.py
│   │
│   ├── schemas/             # Pydantic 模型
│   │   ├── __init__.py
│   │   └── user.py
│   │
│   ├── routers/             # 路由模块
│   │   ├── __init__.py
│   │   ├── users.py
│   │   └── items.py
│   │
│   ├── services/            # 业务逻辑
│   │   ├── __init__.py
│   │   └── user_service.py
│   │
│   ├── middleware/          # 中间件
│   │   └── auth.py
│   │
│   └── utils/               # 工具函数
│       └── security.py
│
├── tests/                   # 测试
│   ├── __init__.py
│   └── test_users.py
│
├── .env                     # 环境变量
├── requirements.txt         # 依赖
└── pytest.ini              # Pytest 配置

模块化路由

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# app/routers/users.py
from fastapi import APIRouter, Depends, HTTPException

router = APIRouter(prefix="/users", tags=["用户"])

@router.get("/")
async def list_users():
    return [{"id": 1, "name": "Alice"}]

@router.get("/{user_id}")
async def get_user(user_id: int):
    return {"id": user_id, "name": f"User {user_id}"}

# app/main.py
from fastapi import FastAPI
from .routers import users, items

app = FastAPI(title="My API")

app.include_router(users.router)
app.include_router(items.router, prefix="/api/v1")

16. 性能优化

异步优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from fastapi import FastAPI
import asyncio
import aiohttp

app = FastAPI()

# 好的异步实践
@app.get("/fetch-data")
async def fetch_data():
    async with aiohttp.ClientSession() as session:
        # 并发请求
        urls = [
            "https://api.example.com/users",
            "https://api.example.com/posts",
            "https://api.example.com/comments"
        ]
        tasks = [session.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        
        # 并发解析
        data = await asyncio.gather(*[r.json() for r in responses])
        
    return {
        "users": data[0],
        "posts": data[1],
        "comments": data[2]
    }

缓存策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from fastapi import FastAPI, Response
from functools import lru_cache
import hashlib
import json

app = FastAPI()

# 内存缓存
cache = {}

@lru_cache(maxsize=128)
def compute_expensive(value: str) -> str:
    """CPU 密集型计算的缓存"""
    return hashlib.sha256(value.encode()).hexdigest()

@app.get("/compute/{value}")
async def compute_endpoint(value: str):
    result = compute_expensive(value)
    return {"result": result}

# HTTP 缓存头
@app.get("/static-data")
async def static_data(response: Response):
    response.headers["Cache-Control"] = "public, max-age=3600"
    return {"data": "static"}

响应压缩

1
2
3
4
5
6
7
8
9
from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()
app.add_middleware(GZipMiddleware, minimum_size=1000)

@app.get("/large-response")
async def large_response():
    return {"data": "x" * 10000}  # 大于 1KB 会自动压缩

附录：常用命令

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 安装
pip install fastapi uvicorn[standard]
pip install python-jose[cryptography]  # JWT
pip install passlib[bcrypt]           # 密码加密
pip install sqlalchemy                # 数据库
pip install "uvicorn[standard]"       # 完整 uvicorn

# 运行
uvicorn main:app --reload             # 开发
uvicorn main:app --workers 4          # 生产

# 测试
pytest
pytest --cov=app                      # 覆盖率

# 代码检查
mypy app                              # 类型检查
black app                             # 格式化
flake8 app                            # 代码规范

学习资源

FastAPI
#FastAPI
FastAPI快速入门自我总结
https://www.furinaluna.top/posts/e8eb82c9.html
作者
Furina
发布于
2026年4月4日
更新于
2026年9月11日
许可协议
BY FURINA