介绍 FastAPI 高性能后端开发，涵盖 ASGI 协议原理、Pydantic 数据验证、依赖注入机制、中间件流程及 Web 安全防护。通过 Hello World、CRUD 实战案例演示 FastAPI 0.100+ 新特性与自动化文档功能，提供路径参数、查询参数处理及常见安全攻击（CSRF、XSS、SQL 注入）的防御方案，帮助开发者快速构建生产级 API 服务。
在现代全栈开发里，FastAPI 已经是很多大厂和初创公司的首选。为什么？
async/await 异步。今天，咱们就用 FastAPI 亲手搭建一个生产级的 API 服务。
首先，你需要安装两个东西：FastAPI 框架本身，和用来运行它的 Web 服务器 uvicorn。
pip install fastapi uvicorn
在深入 FastAPI 之前，我们需要理解它背后的核心协议 —— ASGI（Asynchronous Server Gateway Interface）。
ASGI 是 Python 异步 Web 服务器和应用程序之间的标准接口。你可以把它理解为一座桥梁：
| 协议 | 特性 | 代表框架 |
|---|---|---|
| WSGI | 同步、单线程 | Flask、Django 早期 |
| ASGI | 异步、支持 WebSocket | FastAPI、Django Channels |
ASGI 应用本质上是一个可调用对象（Callable），接收三个参数：
async def application(scope, receive, send):
# scope: 包含请求信息（HTTP 方法、路径、Headers 等）
# receive: 异步函数，用于接收请求体
# send: 异步函数，用于发送响应
await send({'type': 'http.response.start', 'status': 200, 'headers': [(b'content-type', b'text/plain')]})
await send({'type': 'http.response.body', 'body': b'Hello, ASGI!'})
FastAPI 是一个 ASGI 应用框架，它内部使用 Starlette 来处理 ASGI 协议。当你运行 uvicorn main:app 时：
一句话总结：ASGI 让 Python 能够高效处理并发请求，FastAPI 则在此基础上提供了优雅的开发体验。
# main.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"message": "你好，全栈之路！"}
# 启动命令：uvicorn main:app --reload
解释：--reload 是开发神器。你改完代码保存，服务器会自动重启，不需要手动关掉再开。
写 Web 服务最头疼的就是验证用户传过来的数据。万一该传数字的地方传了字符串，程序就崩了。 FastAPI 搭配 Pydantic，一行代码搞定验证：
from pydantic import BaseModel
# 定义一个"商品"的数据结构
class Item(BaseModel):
name: str
price: float
is_offer: bool | None = None
@app.post("/items/")
def create_item(item: Item):
# 如果用户传的 price 不是数字，FastAPI 会自动返回 422 错误
return {"item_name": item.name, "total_price": item.price * 1.2}
当 FastAPI 接收到请求时，Pydantic 会执行以下验证步骤：
@validator 或 @field_validator 装饰的方法from pydantic import BaseModel, field_validator
class User(BaseModel):
username: str
age: int
email: str
@field_validator('username')
@classmethod
def validate_username(cls, v):
if len(v) < 3:
raise ValueError('用户名至少 3 个字符')
if not v.isalnum():
raise ValueError('用户名只能包含字母和数字')
return v
@field_validator('age')
@classmethod
def validate_age(cls, v):
if v < 0 or v > 150:
raise ValueError('年龄必须在 0-150 之间')
return v
# 测试验证
from fastapi import FastAPI
app = FastAPI()
@app.post("/users/")
def create_user(user: User):
return {"message": "用户创建成功", "user": user}
from typing import List
from pydantic import BaseModel
class Address(BaseModel):
city: str
street: str
zipcode: str
class UserWithAddress(BaseModel):
name: str
addresses: List[Address]
# 嵌套模型列表
@app.post("/users-with-address/")
def create_user_with_address(user: UserWithAddress):
return user
一句话总结：Pydantic 在数据进入你的业务逻辑之前就完成所有验证，让你专注于业务而不是防御性编程。
/items/42（指定 ID）。/items/?skip=0&limit=10（翻页、搜索）。@app.get("/users/{user_id}")
def read_user(user_id: int, q: str | None = None):
# user_id 会自动转成整数，q 是可选的搜索关键词
return {"user_id": user_id, "query": q}
依赖注入（Dependency Injection, DI）是一种设计模式，它将对象的创建和管理从使用它的代码中分离出来。
比喻：就像你去餐厅吃饭，不需要自己种菜、做饭，只需要点菜（声明依赖），服务员会把菜端上来（注入依赖）。
FastAPI 的依赖注入系统基于 Python 的函数和类型注解：
from fastapi import Depends, FastAPI
app = FastAPI()
# 定义一个依赖函数
def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
# 使用依赖
@app.get("/items/")
def read_items(commons: dict = Depends(common_parameters)):
return commons
@app.get("/users/")
def read_users(commons: dict = Depends(common_parameters)):
return commons
Depends 标记的依赖from fastapi import Depends, FastAPI
app = FastAPI()
class DatabaseConnection:
def __init__(self):
self.connection = "模拟数据库连接"
def query(self, sql: str):
return f"执行：{sql}"
def close(self):
self.connection = None
def get_db():
db = DatabaseConnection()
try:
yield db # 生成器方式，支持清理操作
finally:
db.close()
@app.get("/data/")
def get_data(db: DatabaseConnection = Depends(get_db)):
result = db.query("SELECT * FROM users")
return {"result": result}
from fastapi import Depends, FastAPI, Header, HTTPException
app = FastAPI()
# 基础依赖：验证 Token
def verify_token(x_token: str = Header(...)):
if x_token != "secret-token":
raise HTTPException(status_code=400, detail="Token 无效")
return x_token
# 子依赖：获取当前用户（依赖 verify_token）
def get_current_user(token: str = Depends(verify_token)):
# 根据 token 查询用户信息
return {"username": "zhangsan", "id": 1}
# 使用依赖链
@app.get("/profile/")
def read_profile(user: dict = Depends(get_current_user)):
return user
一句话总结：依赖注入让代码解耦、可测试、可复用，是 FastAPI 的灵魂特性。
中间件是在请求到达路由处理函数之前和响应返回客户端之前执行的代码。它可以：
请求 -> 中间件 1 -> 中间件 2 -> 路由处理 -> 中间件 2 -> 中间件 1 -> 响应
形象比喻：中间件就像洋葱的层层包裹，请求从外向内穿透，响应从内向外返回。
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import time
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(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
response.headers["X-Process-Time"] = str(process_time)
print(f"[响应] 耗时：{process_time:.4f}s")
return response
# 错误处理中间件
@app.middleware("http")
async def catch_exceptions(request: Request, call_next):
try:
return await call_next(request)
except Exception as e:
return JSONResponse(
status_code=500,
content={"error": str(e)}
)
@app.get("/")
def read_root():
return {"message": "Hello"}
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
app = FastAPI()
# CORS 中间件
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000", "https://example.com"],
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["*"],
)
# GZip 压缩中间件
app.add_middleware(GZipMiddleware, minimum_size=1000)
| 特性 | 中间件 | 依赖注入 |
|---|---|---|
| 执行时机 | 所有请求 | 特定路由 |
| 粒度 | 全局 | 局部 |
| 使用场景 | 日志、CORS、压缩 | 数据库连接、认证 |
| 返回值 | 修改 request/response | 注入到路由函数 |
一句话总结：中间件处理横切关注点，依赖注入提供模块化服务，两者配合让代码更优雅。
Web 安全是每个后端开发者必须掌握的知识。以下是三大常见攻击及防护方法。
攻击原理：诱导用户在已登录的网站上执行非预期的操作。
防护措施：
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
app = FastAPI()
# CSRF Token 验证示例
async def verify_csrf_token(request: Request):
# 从 Header 获取 CSRF Token
csrf_token = request.headers.get("X-CSRF-Token")
# 从 Session/Cookie 获取存储的 Token
stored_token = request.cookies.get("csrf_token")
if not csrf_token or csrf_token != stored_token:
raise HTTPException(status_code=403, detail="CSRF Token 无效")
return True
@app.post("/transfer/")
async def transfer_money(request: Request):
await verify_csrf_token(request)
return {"message": "转账成功"}
最佳实践：
攻击原理：注入恶意脚本到网页中，窃取用户 Cookie 或执行恶意操作。
防护措施：
from fastapi import FastAPI
from pydantic import BaseModel, field_validator
import html
app = FastAPI()
class Comment(BaseModel):
content: str
@field_validator('content')
@classmethod
def sanitize_content(cls, v):
# HTML 转义，防止脚本注入
return html.escape(v)
@app.post("/comments/")
def create_comment(comment: Comment):
# 此时 content 已经被转义，<script> 变成 <script>
return {"content": comment.content}
最佳实践：
攻击原理：通过构造特殊输入，篡改 SQL 查询语句。
防护措施：
from fastapi import FastAPI, HTTPException
import sqlite3
app = FastAPI()
# 错误示例（不要这样做）
@app.get("/users/unsafe/{user_id}")
def get_user_unsafe(user_id: str):
conn = sqlite3.connect("test.db")
cursor = conn.cursor()
# 危险！直接拼接 SQL
cursor.execute(f"SELECT * FROM users WHERE id = {user_id}")
return cursor.fetchone()
# 正确做法：使用参数化查询
@app.get("/users/safe/{user_id}")
def get_user_safe(user_id: int):
conn = sqlite3.connect("test.db")
cursor = conn.cursor()
# 安全：使用占位符，自动转义
cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))
return cursor.fetchone()
最佳实践：
from fastapi import FastAPI
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
app = FastAPI()
# 安全头部中间件
@app.middleware("http")
async def add_security_headers(request, call_next):
response = await call_next(request)
# 防止 XSS
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["X-XSS-Protection"] = "1; mode=block"
# 强制 HTTPS
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
return response
# 限制允许的 Host
app.add_middleware(TrustedHostMiddleware, allowed_hosts=["example.com", "*.example.com"])
一句话总结：安全不是功能，而是底线。永远不信任用户输入，永远使用参数化查询，永远做好输出转义。
FastAPI 0.100 版本是一个重大里程碑，带来了许多重要更新。
FastAPI 0.100+ 默认使用 Pydantic V2，性能提升 5-50 倍：
from pydantic import BaseModel, Field
# Pydantic V2 新特性
class Item(BaseModel):
name: str = Field(min_length=1, max_length=100)
price: float = Field(gt=0, description="必须大于 0")
# V2 新特性：严格模式
model_config = {
"strict": True, # 禁止类型强制转换
"extra": "forbid" # 禁止额外字段
}
from typing import Annotated
from fastapi import FastAPI, Query, Path
app = FastAPI()
# 0.100+ 推荐的新语法
@app.get("/items/{item_id}")
def read_item(
item_id: Annotated[int, Path(title="项目 ID", ge=1)],
q: Annotated[str | None, Query(min_length=3)] = None,
limit: Annotated[int, Query(le=100)] = 10
):
return {"item_id": item_id, "q": q, "limit": limit}
Pydantic V2 使用 Rust 编写的核心验证逻辑，带来：
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
app = FastAPI()
# 自定义验证错误处理
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
errors = []
for error in exc.errors():
errors.append({
"field": " -> ".join(str(x) for x in error["loc"]),
"message": error["msg"],
"type": error["type"]
})
return JSONResponse(
status_code=422,
content={"detail": "数据验证失败", "errors": errors}
)
从旧版本迁移到 0.100+：
# 1. 升级依赖
pip install --upgrade fastapi pydantic
# 2. 检查 Pydantic V2 兼容性
pip install pydantic-v2-migration
python -m pydantic_v2_migration
一句话总结：FastAPI 0.100+ 是一次质的飞跃，Pydantic V2 带来的性能提升让 FastAPI 在生产环境中更加游刃有余。
如果你以前写 API，肯定被写 Swagger 文档折磨过。 在 FastAPI 里，你什么都不用做。服务跑起来后，直接访问：
http://127.0.0.1:8000/docs你会看到一个超级精美的交互式文档。你可以在上面直接点"Try it out"测试你的接口，再也不用开 Postman 了！
咱们来个带增删改查（CRUD）功能的实战案例。
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
# 模拟数据库
todo_db = []
class Todo(BaseModel):
id: int
title: str
completed: bool = False
@app.post("/todos/", status_code=201)
def add_todo(todo: Todo):
todo_db.append(todo)
return {"message": "添加成功", "data": todo}
@app.get("/todos/")
def list_todos():
return todo_db
@app.get("/todos/{todo_id}")
def get_todo(todo_id: int):
for t in todo_db:
if t.id == todo_id:
return t
raise HTTPException(status_code=404, detail="任务找不到了...")
@app.delete("/todos/{todo_id}")
def delete_todo(todo_id: int):
global todo_db
todo_db = [t for t in todo_db if t.id != todo_id]
return {"message": "删除成功"}
async def 别乱用：如果你的函数里没有 await 任何东西（比如没去读文件、没去查数据库），直接写 def 往往更快。FastAPI 会在独立的线程池里跑它，不会阻塞。Depends。它是这款框架的灵魂。需求：
编写一个 API GET /search/，接收参数 keyword（必填，长度至少 2）和 limit（选填，默认 10，最大 50）。
使用 Query 进行参数校验。
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/search/")
def search(
keyword: str = Query(..., min_length=2, description="搜索关键词"),
limit: int = Query(10, le=50, description="返回结果数量")
):
return {"keyword": keyword, "limit": limit, "results": ["模拟数据 1", "模拟数据 2"]}
需求：
编写一个依赖函数 verify_token，检查请求头 x-token 是否为 "fake-super-secret-token"。
如果是，允许访问；否则抛出 400 错误。
将此依赖应用到 GET /protected/ 路由上。
from fastapi import FastAPI, Header, HTTPException, Depends
app = FastAPI()
def verify_token(x_token: str = Header(...)):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="Token 无效")
return x_token
@app.get("/protected/")
def protected_route(token: str = Depends(verify_token)):
return {"message": "验证通过", "token": token}
需求：
编写一个 API POST /upload/，接收一个文件上传。
返回文件的文件名和文件大小。
提示：需要安装 python-multipart。
from fastapi import FastAPI, UploadFile, File
app = FastAPI()
@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
# file.filename 是文件名
# await file.read() 读取内容
content = await file.read()
return {
"filename": file.filename,
"content_type": file.content_type,
"size": len(content)
}
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online