中间件、认证与部署

FastAPI 从开发到上线——中间件处理横切关注点，JWT 保护 API，Docker 打包部署。

FastAPI 生产架构

中间件

"""
FastAPI 中间件：处理请求/响应的横切关注点
"""
from fastapi import FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware
import time
app = FastAPI(title="生产级 API")
# === CORS 跨域 ===
app.add_middleware(
CORSMiddleware,
allow_origins=["https://myapp.com"],  # 生产环境限定域名
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["*"],
allow_credentials=True,
)
# === 请求耗时日志 ===
@app.middleware("http")
async def timing_middleware(request: Request, call_next):
start = time.perf_counter()
response: Response = await call_next(request)
elapsed = time.perf_counter() - start
response.headers["X-Process-Time"] = f"{elapsed:.4f}"
print(f"{request.method} {request.url.path} → {response.status_code} ({elapsed:.3f}s)")
return response
# === 请求 ID 追踪 ===
import uuid
@app.middleware("http")
async def request_id_middleware(request: Request, call_next):
request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
response = await call_next(request)
response.headers["X-Request-ID"] = request_id
return response

JWT 认证

"""
JWT Token 认证
pip install python-jose[cryptography] passlib[bcrypt]
"""
from datetime import datetime, timedelta, timezone
from typing import Optional
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel
# 配置
SECRET_KEY = "your-secret-key-from-env"  # 实际应从环境变量读取
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# 模型
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
username: Optional[str] = None
class User(BaseModel):
username: str
email: str
disabled: bool = False
# 密码哈希
def verify_password(plain: str, hashed: str) -> bool:
return pwd_context.verify(plain, hashed)
def hash_password(password: str) -> str:
return pwd_context.hash(password)
# 创建 Token
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
to_encode = data.copy()
expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=15))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
# 获取当前用户
async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="认证失败",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
# 实际应该从数据库查询
return User(username=username, email=f"{username}@example.com")
# 路由
@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
# 实际应验证数据库中的用户
if form_data.username != "admin":
raise HTTPException(status_code=400, detail="用户名或密码错误")
access_token = create_access_token(
data={"sub": form_data.username},
expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
)
return Token(access_token=access_token, token_type="bearer")
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user

Docker 部署

"""
Docker 部署配置
"""
DOCKERFILE = """
FROM python:3.12-slim
WORKDIR /app
# 依赖先装（利用缓存层）
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制代码
COPY . .
# 非 root 用户运行
RUN useradd -m appuser
USER appuser
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]
"""
DOCKER_COMPOSE = """
version: '3.8'
services:
api:
build: .
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql://user:pass@db:5432/mydb
- SECRET_KEY=${SECRET_KEY}
depends_on:
- db
db:
image: postgres:16
environment:
- POSTGRES_USER=user
- POSTGRES_PASSWORD=pass
- POSTGRES_DB=mydb
volumes:
- pgdata:/var/lib/postgresql/data
volumes:
pgdata:
"""
# Gunicorn 配置（生产推荐）
GUNICORN_CONFIG = """
# gunicorn.conf.py
bind = "0.0.0.0:8000"
workers = 4               # CPU 核数 * 2 + 1
worker_class = "uvicorn.workers.UvicornWorker"
accesslog = "-"
errorlog = "-"
loglevel = "info"
"""

部署选项对比

本章小结

下一章：测试工程化——pytest 与代码质量。