FastAPI 高性能 API：自动文档与异步处理

FreeGuideOnline 最新 2026-06-12

认识 FastAPI 的高性能基因

FastAPI 是当下最快的 Python Web 框架之一，其性能可媲美 Node.js 和 Go。它构建在 Starlette（异步工具）和 Pydantic（数据验证）之上，原生支持异步处理，并具备自动生成交互式 API 文档的能力。本章将带你深入这两个核心特性：自动文档与异步处理，让 API 开发既高效又专业。

自动文档：让 API 自带说明书

FastAPI 最惊艳的特性之一，就是无需任何额外操作，你的 API 就已经拥有一份完整的、可交互的文档。这份文档遵循 OpenAPI 规范，并由 Swagger UI 和 ReDoc 两个业界主流工具渲染。

查看自动生成的文档

当你启动一个 FastAPI 应用后，在浏览器中访问以下地址即可看到文档：

Swagger UI：http://127.0.0.1:8000/docs（可交互操作）
ReDoc：http://127.0.0.1:8000/redoc（更适合阅读与导出）

示例应用：

from fastapi import FastAPI

app = FastAPI()

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

运行后打开 /docs，你会看到一个自动生成的界面，其中列出了路径 /items/{item_id}，并且可以从中直接发送请求进行测试。

文档信息的来源

FastAPI 通过分析以下代码元素自动生成文档：

路径操作函数的签名：如路径参数、查询参数、请求体等。
类型注解：item_id: int 告诉文档该参数为整数。
文档字符串（docstring）：函数的说明文字会显示在文档的描述区域。
Pydantic 模型：用于定义请求体和响应模型的结构，文档会自动展示其字段、类型和示例。

示例：使用 Pydantic 模型丰富文档

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
def create_item(item: Item):
    """
    创建一个新的商品

    此端点返回创建成功的商品信息。
    """
    return item

在 Swagger UI 中，你会看到请求体的 Schema 自动展示，且包含字段的说明和必要标记。

自定义文档元数据

你可以在创建 FastAPI 实例时为文档添加标题、描述、版本等信息，这些会显示在文档的顶端。

app = FastAPI(
    title="我的电商 API",
    description="这是用于管理商品的 API，支持异步高性能处理。",
    version="1.0.0",
)

/docs 和 /redoc 都会使用这些信息。你还可以通过 openapi_tags 给路径分组，让文档结构更清晰。

tags_metadata = [
    {"name": "items", "description": "商品相关操作"},
    {"name": "users", "description": "用户相关操作"},
]

app = FastAPI(openapi_tags=tags_metadata)

@app.get("/items/", tags=["items"])
def list_items():
    return [{"name": "foo"}]

@app.get("/users/", tags=["users"])
def list_users():
    return [{"username": "bar"}]

在 Swagger UI 中，路径会被分组折叠，用户体验更好。

异步处理：榨干单机性能

传统的同步 Python Web 框架（如 Flask）默认使用阻塞式的工作模型，一个请求会占用一个线程或工作进程，当请求中有耗时的 I/O 操作时，线程就会闲置等待，导致吞吐量下降。FastAPI 则是 异步优先 的框架，利用 Python 的 async / await 语法，实现高并发处理。

为什么异步能提高性能？

Python 的异步通过事件循环（event loop）实现协作式多任务。当一个异步函数遇到 await 一个 I/O 操作时，事件循环会挂起该任务，转而执行其他任务，直到 I/O 操作完成再切回来。因此，在等待数据库查询、外部 HTTP 请求、文件读取等 I/O 密集的场景下，一个线程就能处理成千上万的并发连接，而不会因为阻塞导致资源浪费。

在路径操作中使用异步

将路径操作函数声明为 async def 即可启用异步处理。

同步写法（阻塞式）：

@app.get("/sync_items")
def read_sync_items():
    time.sleep(1)   # 模拟耗时同步操作
    return {"message": "同步完成"}

异步写法（非阻塞）：

import asyncio

@app.get("/async_items")
async def read_async_items():
    await asyncio.sleep(1)   # 模拟异步操作，不阻塞事件循环
    return {"message": "异步完成"}

当多个请求同时访问 /sync_items 时，由于 time.sleep 会阻塞整个线程，请求会排队等待；而访问 /async_items 时，asyncio.sleep 会让出控制权，其他请求可以立即得到处理，吞吐量大幅提升。

使用异步数据库驱动

为了真正发挥异步优势，你需要使用支持异步的数据库客户端，例如：

SQLite/PostgreSQL：databases 库或 SQLAlchemy 1.4+ 的异步模式。
MongoDB：motor
Redis：aioredis

示例：使用 databases 执行异步查询

import databases
import sqlalchemy

DATABASE_URL = "sqlite:///./test.db"
database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes", metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
)

engine = sqlalchemy.create_engine(DATABASE_URL)
metadata.create_all(engine)

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

@app.get("/notes/{note_id}")
async def read_note(note_id: int):
    query = notes.select().where(notes.c.id == note_id)
    note = await database.fetch_one(query)
    return note

这里 database.fetch_one 是异步操作，调用时使用 await，不会阻塞事件循环。

注意：不要在异步路径中使用同步阻塞代码

如果你在 async def 函数中执行同步的耗时操作（如 requests.get），它仍然会阻塞事件循环。此时你应将同步任务交给线程池执行，使用 run_in_executor：

import requests
from fastapi.concurrency import run_in_threadpool

@app.get("/external")
async def fetch_external():
    # 错误的做法：直接在异步函数中使用 requests.get（同步阻塞）
    # resp = requests.get("https://example.com")

    # 正确的做法：在线程池中运行
    resp = await run_in_threadpool(requests.get, "https://example.com")
    return {"status": resp.status_code}

这样就能在不阻塞主事件循环的情况下执行同步任务。

自动文档与异步处理结合的最佳实践

一个包含异步数据库操作和自动文档的完整小例子：

from fastapi import FastAPI
from pydantic import BaseModel
import databases, sqlalchemy

# 数据库配置
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()

items = sqlalchemy.Table(
    "items", metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("name", sqlalchemy.String(50)),
    sqlalchemy.Column("price", sqlalchemy.Float),
)

engine = sqlalchemy.create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
metadata.create_all(engine)

# Pydantic 模型
class ItemIn(BaseModel):
    name: str
    price: float

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

# 应用实例
app = FastAPI(title="异步商品API", version="1.0.0")

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

@app.post("/items/", response_model=ItemOut, tags=["items"])
async def create_item(item: ItemIn):
    query = items.insert().values(**item.dict())
    last_id = await database.execute(query)
    return {**item.dict(), "id": last_id}

@app.get("/items/", response_model=list[ItemOut], tags=["items"])
async def list_items():
    query = items.select()
    return await database.fetch_all(query)

运行后访问 /docs，你不但可以查看定义的 ItemIn 和 ItemOut 模型结构，还能直接测试创建和读取商品的操作，所有操作都是异步执行，从容应对高并发。

小结

FastAPI 的高性能不仅来自于框架底层的 Starlette 和异步特性，更体现在它对开发体验的极致优化：

自动文档 无需额外编写，完全来源于代码中的类型注解和模型定义，让 API 的调试、分享和对接变得极其简单。
异步处理 让你用最少的硬件资源处理最大的并发负载，尤其适合 I/O 密集型微服务。

熟练运用这两个特性，你将能快速构建出专业、健壮且高性能的现代 API。在后续的教程中，我们会继续探索依赖注入、安全性、后台任务等高级功能。