FastAPI: FastAPI 简介 — 为什么它是下一代 Python Web 框架
如果说 Flask 是一把灵活的瑞士军刀,Django 是一台功能齐全的越野车,那 FastAPI 就是一辆电动超跑——起步快、能耗低、仪表盘自动生成。
1. 你将学到
- ASGI 与 WSGI 的根本区别:为什么 async 是未来
- FastAPI vs Flask vs Django REST Framework 性能与开发体验对比
- 类型提示(Type Hints)如何驱动自动校验与文档生成
- Starlette + Pydantic 底层架构解读
- PriceTracker 项目全景预览:Alice 将要构建什么
2. Alice 的真实故事
(1) 痛点:同步框架撑不住百万级请求
Alice 是一名后端工程师,正在为电商平台构建 PriceTracker——一个 SaaS 价格追踪 API,需要处理百万级商品价格的实时查询。她最初用 Flask 搭建了原型,但当并发请求突破 1000 QPS 时,同步 WSGI 模型让每个请求排队等待,P99 延迟飙到 3000ms。Bob(前端工程师)抱怨页面加载慢,Charlie(DevOps)说水平扩展成本太高。
(2) FastAPI 的解法
FastAPI 基于 ASGI 异步协议,单进程即可处理数千并发连接,类型提示自动生成 OpenAPI 文档和请求数据校验,Alice 无需手写校验代码和文档。
from fastapi import FastAPI
app = FastAPI()
@app.get("/products/{product_id}")
async def get_product(product_id: int):
# Type hint automatically validates & generates docs
return {"product_id": product_id, "name": "Widget"}
(3) 收益
迁移到 FastAPI 后,PriceTracker 单节点 QPS 从 500 提升到 4000+,P99 延迟降至 50ms,Bob 的前端页面加载快了 5 倍,Charlie 的服务器成本降低 60%。
3. ASGI 与 WSGI:异步是未来
(1) WSGI 的同步瓶颈
WSGI(Web Server Gateway Interface)是 Python Web 的传统标准,每个请求占用一个线程,遇到 I/O 操作(数据库查询、网络请求)时线程阻塞等待。
flowchart LR
Client1[Client 1] -->|Request| WSGI[WSGI Server]
Client2[Client 2] -->|Request| WSGI
Client3[Client 3] -->|Request| WSGI
WSGI -->|Thread 1| DB1[(Database)]
WSGI -->|Thread 2| DB1
WSGI -->|Thread 3 - BLOCKED| DB1
| 维度 | WSGI | ASGI |
|---|---|---|
| 连接模型 | 一请求一线程 | 异步协程,单线程多连接 |
| 并发上限 | 受线程池限制(通常 10-100) | 几乎无上限(协程轻量) |
| I/O 等待 | 阻塞线程 | 非阻塞,切换到其他协程 |
| WebSocket | 不支持 | 原生支持 |
| 典型服务器 | Gunicorn + Flask | Uvicorn + FastAPI |
(2) ASGI 的异步优势
ASGI(Asynchronous Server Gateway Interface)是 WSGI 的异步升级,支持 async/await 语法,单个进程可处理数千并发连接。
import asyncio
import time
# WSGI style - blocks thread
def sync_handler():
time.sleep(1) # Thread blocked for 1 second
return "done"
# ASGI style - non-blocking
async def async_handler():
await asyncio.sleep(1) # Event loop switches to other tasks
return "done"
▶ 示例:同步 vs 异步并发对比
import asyncio
import time
async def fetch_price(product_id: int) -> dict:
# Simulate database I/O latency
await asyncio.sleep(0.1)
return {"product_id": product_id, "price": 9.99}
async def main():
start = time.perf_counter()
# 100 concurrent requests - async finishes in ~0.1s
results = await asyncio.gather(*[fetch_price(i) for i in range(100)])
elapsed = time.perf_counter() - start
print(f"Async: {len(results)} items in {elapsed:.2f}s")
asyncio.run(main())
输出:
执行成功
输出:
Async: 100 items in 0.10s
4. FastAPI vs Flask vs Django DRF
(1) 框架生态定位
flowchart LR
FastAPI[FastAPI] --> Starlette[Starlette ASGI]
Starlette --> Uvicorn[Uvicorn Server]
Uvicorn --> ASGI_Protocol[ASGI Protocol]
FastAPI --> Pydantic[Pydantic V2]
Flask2[Flask] --> Werkzeug[Werkzeug WSGI]
Werkzeug --> Gunicorn[Gunicorn]
Django2[Django DRF] --> Django_Core[Django Core]
| 维度 | FastAPI | Flask | Django DRF |
|---|---|---|---|
| 性能(TechEmpower RPS) | ~40,000 | ~1,200 | ~800 |
| 异步支持 | 原生 async/await | 需扩展 | 有限支持 |
| 自动文档 | OpenAPI 自动生成 | 需 Flask-RESTX | 需 drf-spectacular |
| 类型校验 | Pydantic 自动校验 | 手动校验 | Serializer 手动定义 |
| 学习曲线 | 低(类型提示即文档) | 低 | 高 |
| 项目规模 | 中小型 API 服务 | 小型服务 | 大型全栈项目 |
(2) 为什么 FastAPI 更适合 PriceTracker
| PriceTracker 需求 | FastAPI 优势 | Flask 劣势 |
|---|---|---|
| 百万级查询 QPS | 异步协程高并发 | 同步阻塞低并发 |
| 实时价格推送 WebSocket | 原生支持 | 不支持 |
| 自动 API 文档给 Bob | OpenAPI 自动生成 | 需额外配置 |
| 请求数据校验 | Pydantic 自动 | 手写校验装饰器 |
| JWT 认证 | OAuth2 工具内置 | 需第三方库 |
▶ 示例:相同 API 的三种写法对比
# === FastAPI: Type hints = auto validation + docs ===
from fastapi import FastAPI
from pydantic import BaseModel
class Product(BaseModel):
name: str
price: float
app = FastAPI()
@app.post("/products")
async def create_product(product: Product):
return product # Auto validated, auto documented
输出:
# 函数定义成功
# === Flask: Manual validation, no auto docs ===
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route("/products", methods=["POST"])
def create_product():
data = request.get_json()
if not data or "name" not in data or "price" not in data:
return jsonify({"error": "Invalid data"}), 400
if not isinstance(data["price"], (int, float)):
return jsonify({"error": "Price must be number"}), 400
return jsonify(data)
5. 类型提示驱动一切
(1) 类型提示的三重价值
FastAPI 利用 Python 类型提示同时实现三件事:数据校验、序列化/反序列化、OpenAPI 文档生成。
| 类型提示功能 | 传统框架 | FastAPI |
|---|---|---|
| 数据校验 | 手写 if/else | Pydantic 自动 |
| JSON 序列化 | 手动 json.dumps |
model_dump() 自动 |
| API 文档 | 手写 Swagger YAML | OpenAPI 自动生成 |
| IDE 自动补全 | 无 | 完整类型推导 |
▶ 示例:类型提示自动校验
from fastapi import FastAPI, Query
from typing import Optional
app = FastAPI()
@app.get("/prices")
async def search_prices(
min_price: float = Query(0.0, ge=0, description="Minimum price in USD"),
max_price: float = Query(999999.0, le=999999, description="Maximum price in USD"),
category: Optional[str] = Query(None, max_length=50),
):
return {"min_price": min_price, "max_price": max_price, "category": category}
输出:
# 函数定义成功
▶ 示例:自动生成的 OpenAPI 文档
# After defining the endpoint above, visit:
# http://localhost:8000/docs -> Swagger UI
# http://localhost:8000/redoc -> ReDoc
# http://localhost:8000/openapi.json -> Raw OpenAPI schema
输出(
/openapi.json节选):
{
"paths": {
"/prices": {
"get": {
"summary": "Search Prices",
"parameters": [
{"name": "min_price", "in": "query", "schema": {"type": "number", "minimum": 0.0}}
]
}
}
}
}
(2) Pydantic 的角色
Pydantic V2 是 FastAPI 的数据校验引擎,用 Rust 重写核心,比 V1 快 5-50 倍。
| 特性 | Pydantic V1 | Pydantic V2 |
|---|---|---|
| 核心引擎 | Python | Rust(pydantic-core) |
| 校验速度 | 基准 | 5-50x 更快 |
| 校验器 | @validator |
@field_validator/@model_validator |
| 配置 | class Config |
model_config = ConfigDict(...) |
| 序列化 | .dict() |
.model_dump() |
6. 底层架构:Starlette + Pydantic
(1) FastAPI 的三层架构
FastAPI 本身是薄封装层,核心能力来自 Starlette(ASGI 框架)和 Pydantic(数据校验)。
| 层级 | 组件 | 职责 |
|---|---|---|
| 应用层 | FastAPI | 路由注册、依赖注入、OpenAPI 生成 |
| ASGI 层 | Starlette | 中间件、请求/响应处理、WebSocket |
| 校验层 | Pydantic | 数据校验、序列化、类型转换 |
| 服务层 | Uvicorn | ASGI 服务器,事件循环管理 |
▶ 示例:FastAPI 直接使用 Starlette 功能
from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware
from starlette.responses import JSONResponse
app = FastAPI()
# Starlette middleware works seamlessly with FastAPI
app.add_middleware(CORSMiddleware, allow_origins=["*"])
# Starlette response classes work too
@app.get("/health")
async def health_check():
return JSONResponse({"status": "healthy"})
输出:
# 函数定义成功
▶ 示例:Pydantic 模型在 FastAPI 中的流转
from fastapi import FastAPI
from pydantic import BaseModel, Field
class PriceCreate(BaseModel):
product_id: int = Field(gt=0)
price: float = Field(gt=0, description="Price in USD")
currency: str = Field(default="USD", max_length=3)
app = FastAPI()
@app.post("/prices")
async def create_price(data: PriceCreate):
# data is already validated & parsed by Pydantic
validated = data.model_dump()
return {"status": "created", "data": validated}
输出:
# 函数定义成功
7. PriceTracker 项目全景预览
(1) Alice 要构建什么
PriceTracker 是一个 SaaS 价格追踪 API,核心功能:
| 功能模块 | API 端点 | 说明 |
|---|---|---|
| 商品管理 | /products CRUD |
百万级商品数据 |
| 价格追踪 | /prices CRUD |
实时价格查询与推送 |
| 用户认证 | /auth/login, /auth/register |
JWT 双令牌认证 |
| 订阅计划 | Free/Pro/Enterprise | SaaS 多租户权限 |
| 批量导入 | /import/csv |
千行 CSV 异步导入 |
| 实时推送 | WebSocket /ws/prices |
价格变动即时通知 |
flowchart LR
Bob[Bob - Frontend] -->|HTTP/WebSocket| API[PriceTracker API]
Charlie[Charlie - DevOps] -->|Monitor| API
API -->|Query| DB[(PostgreSQL)]
API -->|Cache| Redis[(Redis)]
API -->|Task| Celery[Celery Worker]
Celery -->|Scrape| External[External Sites]
▶ 示例:PriceTracker 最小可运行版本
from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional
app = FastAPI(title="PriceTracker API", version="0.1.0")
class ProductCreate(BaseModel):
name: str = Field(max_length=200)
category: str = Field(max_length=100)
base_price: float = Field(gt=0, description="Base price in USD")
class ProductResponse(BaseModel):
id: int
name: str
category: str
base_price: float
PRODUCTS_DB: dict[int, dict] = {}
_counter = 0
@app.post("/products", response_model=ProductResponse)
async def create_product(product: ProductCreate):
global _counter
_counter += 1
record = {"id": _counter, **product.model_dump()}
PRODUCTS_DB[_counter] = record
return record
@app.get("/products/{product_id}", response_model=ProductResponse)
async def get_product(product_id: int):
if product_id not in PRODUCTS_DB:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Product not found")
return PRODUCTS_DB[product_id]
输出:
# 函数定义成功
8. 综合示例
FastAPI 的核心优势是类型提示驱动的自动校验与文档生成。下面演示一个完整的价格查询端点,整合路径参数、Pydantic 模型与响应模型。
from fastapi import FastAPI
from pydantic import BaseModel, Field
app = FastAPI(title="PriceTracker Demo")
class PriceResponse(BaseModel):
product_id: int = Field(gt=0)
product_name: str
price: float = Field(gt=0)
currency: str = "USD"
PRICES_DB: dict[int, dict] = {
1: {"product_id": 1, "product_name": "Widget", "price": 9.99},
2: {"product_id": 2, "product_name": "Gadget", "price": 24.50},
}
@app.get("/products/{product_id}", response_model=PriceResponse)
async def get_price(product_id: int):
if product_id not in PRICES_DB:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Product not found")
return PRICES_DB[product_id]
输出:
GET /products/1 → {"product_id":1,"product_name":"Widget","price":9.99,"currency":"USD"}
GET /products/99 → 404 Not Found
❓ 常见问题
📖 小节
- FastAPI 基于 ASGI 异步协议,单进程可处理数千并发,性能远超 WSGI 框架
- 类型提示同时驱动数据校验、序列化和 OpenAPI 文档生成,减少 70% 样板代码
- FastAPI = Starlette(ASGI 框架)+ Pydantic(数据校验)的薄封装,各层可独立使用
- Pydantic V2 用 Rust 重写核心,比 V1 快 5-50 倍,是 FastAPI 性能的关键
- PriceTracker 项目将贯穿 25 课,从零构建到生产部署,涵盖百万级 SaaS 场景
📝 作业
- 基础题(难度⭐):安装 FastAPI 和 Uvicorn,创建一个返回
{"message": "Hello PriceTracker"}的 GET 端点,并用uvicorn启动。提示:pip install fastapi uvicorn - 进阶题(难度⭐⭐):为端点添加路径参数
name,返回{"message": "Hello, {name}"},并在浏览器访问/docs查看自动文档。提示:@app.get("/hello/{name}") - 挑战题(难度⭐⭐⭐):创建一个 Pydantic 模型
PriceInput,包含product_name: str和price: float(必须 > 0),用 POST 端点接收数据并返回验证后的结果。提示:继承BaseModel,用Field(gt=0)
---|