FastAPI Web 开发

Python 最快的 Web 框架——类型安全、自动文档、异步原生。

FastAPI 架构

快速入门

"""
FastAPI 入门
pip install fastapi uvicorn
"""
from fastapi import FastAPI, HTTPException, Depends, Query
from pydantic import BaseModel, Field
from datetime import datetime
app = FastAPI(
title="用户管理 API",
description="一个完整的 RESTful API 示例",
version="1.0.0",
)
# === 数据模型 ===
class UserCreate(BaseModel):
"""创建用户请求体"""
name: str = Field(..., min_length=2, max_length=50, examples=["张三"])
email: str = Field(..., examples=["zhangsan@example.com"])
age: int = Field(..., ge=0, le=150, examples=[25])
class UserResponse(BaseModel):
"""用户响应体"""
id: int
name: str
email: str
age: int
created_at: datetime
class UserUpdate(BaseModel):
"""更新用户请求体"""
name: str | None = None
email: str | None = None
age: int | None = Field(None, ge=0, le=150)
# === 模拟数据库 ===
fake_db: dict[int, dict] = {}
next_id = 1
# === 路由 ===
@app.get("/")
async def root():
"""首页"""
return {"message": "Welcome to User API", "docs": "/docs"}
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
"""创建用户"""
global next_id
user_data = {
"id": next_id,
**user.model_dump(),
"created_at": datetime.now(),
}
fake_db[next_id] = user_data
next_id += 1
return user_data
@app.get("/users", response_model=list[UserResponse])
async def list_users(
skip: int = Query(0, ge=0, description="跳过条数"),
limit: int = Query(10, ge=1, le=100, description="每页条数"),
):
"""获取用户列表"""
users = list(fake_db.values())
return users[skip: skip + limit]
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
"""获取单个用户"""
if user_id not in fake_db:
raise HTTPException(status_code=404, detail="用户不存在")
return fake_db[user_id]
@app.put("/users/{user_id}", response_model=UserResponse)
async def update_user(user_id: int, user: UserUpdate):
"""更新用户"""
if user_id not in fake_db:
raise HTTPException(status_code=404, detail="用户不存在")
update_data = user.model_dump(exclude_unset=True)
fake_db[user_id].update(update_data)
return fake_db[user_id]
@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
"""删除用户"""
if user_id not in fake_db:
raise HTTPException(status_code=404, detail="用户不存在")
del fake_db[user_id]
return {"message": f"用户 {user_id} 已删除"}
# 启动: uvicorn main:app --reload
# 文档: http://localhost:8000/docs

依赖注入

"""
FastAPI 依赖注入
"""
from fastapi import Depends, Header, HTTPException
# === 通用依赖 ===
async def verify_token(authorization: str = Header(...)):
"""验证 Token"""
if not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="无效的认证头")
token = authorization[7:]
if token != "secret-token":
raise HTTPException(status_code=401, detail="无效的 Token")
return token
async def get_current_user(token: str = Depends(verify_token)):
"""获取当前用户"""
# 实际项目中解析 JWT Token
return {"id": 1, "name": "当前用户", "role": "admin"}
# 使用依赖
# @app.get("/profile")
# async def profile(user = Depends(get_current_user)):
#     return user
# === 数据库依赖 ===
class Database:
"""模拟数据库连接"""
def __init__(self):
self.connected = False
async def connect(self):
self.connected = True
async def disconnect(self):
self.connected = False
async def get_db():
"""数据库会话依赖"""
db = Database()
await db.connect()
try:
yield db  # 请求处理期间提供 db
finally:
await db.disconnect()  # 请求结束后关闭
# @app.get("/data")
# async def get_data(db: Database = Depends(get_db)):
#     return {"connected": db.connected}
print("FastAPI 依赖注入示例准备就绪")

中间件与错误处理

"""
中间件和错误处理
"""
import time
from fastapi import Request
from fastapi.responses import JSONResponse
# === 中间件 ===
# @app.middleware("http")
async def timing_middleware(request: Request, call_next):
"""请求计时中间件"""
start = time.time()
response = await call_next(request)
duration = time.time() - start
response.headers["X-Process-Time"] = f"{duration:.4f}"
return response
# === 全局异常处理 ===
class AppError(Exception):
def __init__(self, code: int, message: str):
self.code = code
self.message = message
# @app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError):
return JSONResponse(
status_code=exc.code,
content={"error": exc.message, "code": exc.code},
)
# === 项目结构建议 ===
PROJECT_STRUCTURE = """
my_api/
├── main.py              # 入口
├── config.py            # 配置
├── models/              # Pydantic 模型
│   ├── __init__.py
│   └── user.py
├── routers/             # 路由
│   ├── __init__.py
│   └── users.py
├── services/            # 业务逻辑
│   ├── __init__.py
│   └── user_service.py
├── dependencies.py      # 依赖
├── middleware.py         # 中间件
└── requirements.txt
"""
print(PROJECT_STRUCTURE)

本章小结

下一章：测试与工程化——pytest、类型注解、代码质量。