FastAPI 基于 ASGI 协议，支持异步编程和类型提示。介绍环境搭建、Pydantic 数据验证、依赖注入、中间件机制及 Web 安全防护（CSRF、XSS、SQL 注入）。包含实战案例如待办事项 API 构建，并涵盖 FastAPI 0.100+ 新特性如 Pydantic V2 支持。适合希望构建高性能 Python 后端服务的开发者。
在现代全栈开发里，FastAPI 已经是很多大厂和初创公司的首选。
async/await 异步编程。安装 FastAPI 框架及 Web 服务器 uvicorn。
pip install fastapi uvicorn
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 时：
# main.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"message": "你好，全栈之路！"}
# 启动命令：uvicorn main:app --reload
解释：--reload 是开发神器。修改代码保存后，服务器会自动重启。
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):
if item.price < 0:
raise ValueError("价格不能为负")
return {"item_name": item.name, "total_price": item.price * 1.2}
@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
@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
/items/42（指定 ID）。/items/?skip=0&limit=10（翻页、搜索）。@app.get("/users/{user_id}")
def read_user(user_id: int, q: str | None = None):
return {"user_id": user_id, "query": q}
依赖注入（Dependency Injection, DI）是一种设计模式，将对象的创建和管理从使用它的代码中分离出来。
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
Depends 标记的依赖from fastapi import Depends, 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()
def verify_token(x_token: str = Header(...)):
if x_token != "secret-token":
raise HTTPException(status_code=400, detail="Token 无效")
return x_token
def get_current_user(token: str = Depends(verify_token)):
return {"username": "zhangsan", "id": 1}
@app.get("/profile/")
def read_profile(user: dict = Depends(get_current_user)):
return user
中间件是在请求到达路由处理函数之前和响应返回客户端之前执行的代码。它可以记录日志、添加响应头、处理跨域（CORS）、认证授权、压缩响应等。
请求 -> 中间件 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()
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000", "https://example.com"],
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["*"],
)
app.add_middleware(GZipMiddleware, minimum_size=1000)
防护措施：
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
app = FastAPI()
async def verify_csrf_token(request: Request):
csrf_token = request.headers.get("X-CSRF-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": "转账成功"}
防护措施：
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):
return html.escape(v)
@app.post("/comments/")
def create_comment(comment: Comment):
return {"content": comment.content}
防护措施：永远使用参数化查询/预编译语句。
from fastapi import FastAPI, HTTPException
import sqlite3
app = FastAPI()
@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)
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
return response
app.add_middleware(TrustedHostMiddleware, allowed_hosts=["example.com", "*.example.com"])
FastAPI 0.100+ 默认使用 Pydantic V2，性能提升显著。
from pydantic import BaseModel, Field
class Item(BaseModel):
name: str = Field(min_length=1, max_length=100)
price: float = Field(gt=0, description="必须大于 0")
model_config = {
"strict": True,
"extra": "forbid"
}
from typing import Annotated
from fastapi import FastAPI, Query, Path
app = FastAPI()
@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}
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})
FastAPI 自动生成交互式文档。服务跑起来后，访问：
http://127.0.0.1:8000/docsfrom 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": "删除成功"}
await，直接写 def 往往更快。Depends。编写一个 API GET /search/，接收参数 keyword（必填，长度至少 2）和 limit（选填，默认 10，最大 50）。
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'。
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/，接收一个文件上传。
from fastapi import FastAPI, UploadFile, File
app = FastAPI()
@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
content = await file.read()
return {
"filename": file.filename,
"content_type": file.content_type,
"size": len(content)
}
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。 在线工具，HTML转Markdown在线工具，online
通过删除不必要的空白来缩小和压缩JSON。 在线工具，JSON 压缩在线工具，online