前言

FastAPI 是一个基于 Python 3.7+ 的现代化 Web 框架，用于构建高性能的 API 应用。它最大的特点是将类型提示与 Web 开发完美结合，让开发者能够用简洁的 Python 代码快速构建功能强大的 API。

使用上，先创建 FastAPI 实例，然后使用装饰器来定义路由。比如 @app.get("/") 定义请求的端点，通过与 Pydantic 的深度集成，可以定义数据模型类来自动处理请求验证、响应序列化和文档生成。

此外，FastAPI 会根据代码自动生成交互式的 API 文档，包括 Swagger UI 和 ReDoc 两种格式，不需要额外的工作就能获得完整的 API 文档。框架还内置了对异步编程的支持，能够处理高并发场景，性能表现非常出色。

ASGI 服务器

在开始教程前，先来了解一下 ASGI 服务器这个重要的基础概念。

ASGI（Asynchronous Server Gateway Interface）是 Python Web 应用和 Web 服务器之间的异步接口标准，可以理解为传统 WSGI(Web Server Gateway Interface) 的现代化异步版本。在传统的 Web 开发中，Python 使用 WSGI 标准来连接 Web 应用和服务器，但 WSGI 是同步的，无法很好地处理异步需求，比如 WebSocket 连接、服务器推送事件、长轮询等。ASGI 的出现就是为了解决这些问题，它原生支持异步编程模式，能够处理 HTTP、WebSocket 和其他协议的异步通信。

Uvicorn 是最流行的 ASGI 服务器之一，它基于 uvloop 和 httptools 构建，提供了极高的性能。Uvicorn 负责启动服务器、监听端口、处理网络连接，而 FastAPI 则专注于路由处理、数据验证、业务逻辑等应用层面的工作。

除了 Uvicorn，还有其他 ASGI 服务器选择，比如 Hypercorn、Daphne 等，但由于 Uvicorn 出色的性能以及与 FastAPI 的良好兼容性成为了首选。

快速上手

1. 安装 FastAPI

1
2
pip install fastapi
pip install "uvicorn[standard]"  # ASGI 服务器

2. 简单示例

创建一个名为 main.py 的文件：

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

# 创建 FastAPI 实例
app = FastAPI()

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

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

启动服务：

1
uvicorn main:app --reload

程序说明：

  • main:app 表示从 main.py 文件中导入名为 app 的 FastAPI 实例
  • --reload 参数启用热重载功能，当代码发生变化时会自动重启服务器（仅在开发环境使用）
  • 服务默认运行在 http://127.0.0.1:8000

访问方式：

1
2
3
4
5
6
7
8
❯ curl http://localhost:8000
{"Hello":"World"}

❯ curl http://127.0.0.1:8000/items/123
{"item_id":123,"q":null}

❯ curl http://127.0.0.1:8000/items/123\?q\=test
{"item_id":123,"q":"test"}
  • 访问 http://127.0.0.1:8000/docs 查看自动生成的 Swagger UI 文档
  • 访问 http://127.0.0.1:8000/redoc 查看 ReDoc 格式的文档

代码解析：

  • @app.get("/") 是路由装饰器，定义了一个处理 GET 请求的端点
  • {item_id} 是路径参数，FastAPI 会自动将 URL 中的值传递给函数参数
  • item_id: int 类型注解告诉 FastAPI 这个参数应该是整数，如果传入非整数会自动返回错误
  • q: str = None 定义了一个可选的查询参数，默认值为 None

3. 实战示例

创建一个更实用的示例，这个示例将捕获请求并记录详细信息。这对于调试和监控 API 非常有用：

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
from fastapi import FastAPI, Request
from datetime import datetime
import uvicorn

app = FastAPI(
    title="请求监控 API",
    description="捕获并展示所有 HTTP 请求的详细信息",
    version="1.0.0"
)

@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def capture_all_requests(request: Request, path: str = ""):
    """
    捕获所有进入的 HTTP 请求并返回详细信息
    
    - path: 请求路径（支持任意深度的路径）
    - request: 请求对象，包含所有请求相关信息
    """
    # 获取请求时间
    timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    
    # 获取请求体（如果有）
    body = await request.body()
    body_str = body.decode('utf-8') if body else ""
    
    # 构建请求信息
    request_info = {
        "timestamp": timestamp,
        "method": request.method,
        "url": str(request.url),
        "path": path,
        "client": {
            "host": request.client.host,
            "port": request.client.port
        },
        "headers": dict(request.headers),
        "query_params": dict(request.query_params),
        "body": body_str
    }
    
    # 打印请求信息到控制台
    print("\n" + "="*50)
    print(f"📝 新请求 | {timestamp}")
    print(f"📌 方法: {request.method}")
    print(f"🔗 URL: {request.url}")
    print(f"👤 客户端: {request.client.host}:{request.client.port}")
    print(f"🔍 查询参数: {dict(request.query_params)}")
    if body_str:
        print(f"📦 请求体: {body_str}")
    print(f"📋 请求头:")
    for key, value in request.headers.items():
        print(f"   {key}: {value}")
    print("="*50)
    
    return {
        "status": "success",
        "message": "请求已记录",
        "request_info": request_info
    }

if __name__ == "__main__":
    # 启动服务器
    uvicorn.run(
        app,
        host="127.0.0.1",
        port=8000,
        reload=True  # 开发模式下启用热重载
    )

这个示例主要通过 @app.api_route 装饰器来捕获所有 HTTP 方法的请求。

4. 生命周期

应用生命周期是指一个 Web 应用从启动到关闭的完整过程。在这个过程中，应用需要进行各种初始化和清理工作，比如：

  • 启动时：连接数据库、加载配置、预热缓存、初始化机器学习模型等
  • 运行时：处理用户请求、维持连接状态等
  • 关闭时：关闭数据库连接、保存状态、清理临时文件等

FastAPI 通过 lifespan 事件提供了完整的解决方案。

使用示例：

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

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动阶段
    print("应用启动")
    # 初始化工作
    yield
    # 关闭阶段
    print("应用关闭")
    # 清理工作

app = FastAPI(lifespan=lifespan)

当服务被 ctrl + c 中断或者运行 kill <pid> 关闭服务时，会触发关闭阶段的代码。但如果是强制 kill（如 kill -9 <pid>），则不会触发这一阶段。

uvicorn 配置

uvicorn 是一个轻量级的 ASGI 服务器，提供了多种配置选项来满足不同的部署需求。

启动方式

1
2
3
4
5
6
7
8
# 方式一：命令行启动
uvicorn main:app --reload

# 方式二：代码中启动
import uvicorn

if __name__ == "__main__":
    uvicorn.run("main:app", reload=True)

常用参数

1
2
3
4
5
6
7
8
9
10
11
uvicorn.run(
    "main:app",                # 应用程序路径 ("文件名:FastAPI实例名")
    host="127.0.0.1",         # 监听地址
    port=8000,                # 监听端口
    reload=True,              # 热重载，开发环境使用
    workers=4,                # 工作进程数，一般为CPU核心数的2-4倍
    log_level="info",         # 日志级别: critical, error, warning, info, debug, trace
    access_log=True,          # 是否显示访问日志
    limit_concurrency=1000,   # 并发连接数限制
    timeout_keep_alive=5,     # keep-alive 连接超时时间(秒)
)

命令行参数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 基本用法
uvicorn main:app --reload

# 指定主机和端口
uvicorn main:app --host 0.0.0.0 --port 80

# 生产环境配置
uvicorn main:app --workers 4 --host "::" --port 80 --proxy-headers --forwarded-allow-ips='*'

# SSL/HTTPS 配置
uvicorn main:app --ssl-keyfile=./key.pem --ssl-certfile=./cert.pem

# 日志配置
uvicorn main:app --log-level debug --access-log

环境变量

uvicorn 也支持通过环境变量配置：

1
2
3
4
5
6
7
# 设置环境变量
export UVICORN_HOST=0.0.0.0
export UVICORN_PORT=80
export UVICORN_WORKERS=4

# 启动服务器（将使用环境变量中的配置）
uvicorn main:app

Gunicorn 配置

Gunicorn（Green Unicorn）是一个 Python WSGI HTTP 服务器，广泛用于生产环境中部署 Django、Flask 等基于 WSGI 的应用。

安装方式：

1
pip install gunicorn

与 Uvicorn 对比：

特征 Gunicorn Uvicorn
服务器模型 同步模型 异步模型
工作方式 多进程/多线程 事件循环
性能 适用于 CPU 密集型任务 适用于 I/O 密集型任务
内存使用 相对较高 效率更高

Gunicorn + Uvicorn

Gunicorn + Uvicorn Worker 是一种将传统进程管理与现代异步性能相结合的部署方案。Gunicorn 作为主控制器负责进程管理、监控和重启，而 UvicornWorker 提供异步处理能力。

这种方案特别适合需要高性能异步处理但又要求生产级稳定性的场景，是目前部署 FastAPI 应用的主流选择之一。

1
2
3
4
5
6
7
8
9
10
11
12
# 架构示意
┌─────────────────┐
│   Gunicorn      │  ← 主进程，负责管理
│   Master        │
└─────────────────┘
         │
    ┌────┴────┬────────┬────────┐
    │         │        │        │
┌───▼───┐ ┌──▼───┐ ┌──▼───┐ ┌──▼───┐
│Worker1│ │Worker2│ │Worker3│ │Worker4│  ← UvicornWorker 进程
│(ASGI) │ │(ASGI) │ │(ASGI) │ │(ASGI) │
└───────┘ └──────┘ └──────┘ └──────┘

使用示例

创建一个 app.py 文件：

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

app = FastAPI(title="Simple Gunicorn + Uvicorn Demo")

@app.get("/")
def root():
    return {
        "message": "Hello from Gunicorn + Uvicorn!",
        "worker_pid": os.getpid()  # 显示当前进程ID
    }

@app.get("/health")
def health():
    return {"status": "ok", "pid": os.getpid()}

if __name__ == "__main__":
    # 开发环境直接用 uvicorn
    uvicorn.run("app:app", host="127.0.0.1", port=8000, reload=True)

启动服务：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 方式1: 开发环境 - 直接用 uvicorn
python app.py

# 方式2: 生产环境 - 使用 Gunicorn + UvicornWorker
gunicorn app:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000

# 方式3: 带更多参数的生产环境启动
gunicorn app:app \
    --workers 4 \
    --worker-class uvicorn.workers.UvicornWorker \
    --bind 0.0.0.0:8000 \
    --timeout 120 \
    --access-logfile - \
    --error-logfile -

测试

1
2
3
4
5
# 测试根路径
curl http://localhost:8000/

# 测试健康检查
curl http://localhost:8000/health

NOTE: 和 Uvicorn 相比，Gunicorn 是进程隔离的服务，且支持很多服务管理策略，包括负载均衡和定期重启等功能，但对资源的消耗也更多。个人实测发现，Gunicorn 对资源消耗也比较大，对 CPU 利用率低，后续找时间再专门测试分析看看。目前的个人体会：框架越简洁，支持的功能越少，并发速度也越快。反之亦然，因此个人使用 ray 时，也经常发现性能卡点。通常，可以更多地借助 Nginx 之类成熟的负载均衡工具，来提高扩展并发能力。

其他重要特性

FastAPI 除了基本的路由和生命周期管理外，还提供了许多强大的特性来帮助开发者构建健壮的 API 服务。

Pydantic 模型

Pydantic 是 FastAPI 的核心组件之一，它提供了数据验证、序列化和文档生成的能力。通过定义 Pydantic 模型，你可以：

  • 自动数据验证：确保接收到的数据符合预期格式
  • 类型安全：利用 Python 类型提示进行静态检查
  • 自动文档生成：API 文档会自动包含模型结构
  • 数据转换：自动处理 JSON 到 Python 对象的转换

代码示例：

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
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

# 定义数据模型
class Item(BaseModel):
    name: str
    price: float
    is_offer: Optional[bool] = None

class User(BaseModel):
    username: str
    email: str
    full_name: Optional[str] = None

# 使用模型
@app.post("/items/")
def create_item(item: Item):
    return item

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

使用 Pydantic 模型的好处：

  • 如果客户端发送的数据不符合模型定义（比如 price 不是数字），FastAPI 会自动返回 400 错误
  • 模型字段会自动出现在 API 文档中，包括类型信息和是否必填
  • 支持默认值、可选字段等高级特性

响应模型格式

通过 response_model 参数，你可以精确控制 API 返回的数据格式。这在以下场景特别有用：

  • 数据过滤：只返回部分字段，隐藏敏感信息
  • 格式标准化：确保所有 API 返回一致的数据结构
  • 文档准确性：让 API 文档准确反映实际返回的数据
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
from pydantic import BaseModel
from typing import List

app = FastAPI()

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

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

@app.post("/items/", response_model=ItemResponse)
def create_item(item: Item):
    return ItemResponse(name=item.name, price=item.price, id=1)

@app.get("/items/", response_model=List[ItemResponse])
def read_items():
    return [
        ItemResponse(name="Item 1", price=10.0, id=1),
        ItemResponse(name="Item 2", price=20.0, id=2)
    ]

响应模型的优势：

  • 类型安全：确保返回的数据结构正确
  • 自动序列化：Python 对象自动转换为 JSON
  • 文档生成：API 文档会显示准确的响应格式
  • 数据验证：即使是返回数据也会经过验证

异常处理

合理的异常处理是生产环境 API 的必备特性。FastAPI 通过 HTTPException 提供了标准化的错误处理机制：

1
2
3
4
5
6
7
8
9
10
11
from fastapi import FastAPI, HTTPException

app = FastAPI()

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

@app.get("/items/{item_id}")
def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}

异常处理的最佳实践：

  • 使用标准 HTTP 状态码：404 表示未找到，400 表示请求错误等
  • 提供有意义的错误信息：帮助客户端理解和处理错误
  • 保持错误格式一致：所有 API 的错误响应格式应该统一

完整示例

以下是一个综合运用上述特性的完整 CRUD API 示例，展示了如何构建一个生产就绪的 API 服务：

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
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional

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

# 数据模型
class Item(BaseModel):
    id: Optional[int] = None
    name: str
    price: float
    description: Optional[str] = None

# 模拟数据库
items_db = []

@app.get("/")
def root():
    return {"message": "Welcome to my API"}

@app.get("/items/", response_model=List[Item])
def get_items():
    return items_db

@app.get("/items/{item_id}", response_model=Item)
def get_item(item_id: int):
    for item in items_db:
        if item.id == item_id:
            return item
    raise HTTPException(status_code=404, detail="Item not found")

@app.post("/items/", response_model=Item)
def create_item(item: Item):
    item.id = len(items_db) + 1
    items_db.append(item)
    return item

@app.put("/items/{item_id}", response_model=Item)
def update_item(item_id: int, updated_item: Item):
    for i, item in enumerate(items_db):
        if item.id == item_id:
            updated_item.id = item_id
            items_db[i] = updated_item
            return updated_item
    raise HTTPException(status_code=404, detail="Item not found")

@app.delete("/items/{item_id}")
def delete_item(item_id: int):
    for i, item in enumerate(items_db):
        if item.id == item_id:
            del items_db[i]
            return {"message": "Item deleted"}
    raise HTTPException(status_code=404, detail="Item not found")

API 文档

FastAPI 最大的亮点之一是自动生成交互式 API 文档。无需额外配置，启动应用后就可以访问：

  • Swagger UI: http://127.0.0.1:8000/docs

    • 提供交互式界面，可以直接测试 API
    • 显示所有端点、参数、响应格式
    • 支持在线调试和测试

  • ReDoc: http://127.0.0.1:8000/redoc

    • 提供更加美观的文档展示
    • 适合作为 API 说明文档分享给其他开发者
    • 响应式设计，支持移动设备

这些特性使得 FastAPI 不仅适合快速原型开发，也能够支撑大型生产环境的 API 服务需求。