FastAPI 以其现代的异步优先方法彻底改变了 Python API 开发。我花了大量时间研究 FastAPI，我将分享在生产环境中行之有效的高级模式。
依赖注入
FastAPI 中的依赖注入提供了清晰的关注点分离和高效的资源管理。以下是数据库依赖项的实际实现：
from fastapi import Depends
from sqlalchemy.orm import Session
from contextlib import contextmanager
class Database:
def __init__(self):
self.session = None
@contextmanager
def get_session(self):
session = Session()
try:
yield session
finally:
session.close()
db = Database()
async def get_db():
with db.get_session() as session:
yield session
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_db)):
return db.query(User).filter(User.id == user_id).first()
响应缓存
实现 Redis 缓存可显著提高 API 性能。以下是一个强大的缓存实现：
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
import pickle
class CustomRedisBackend(RedisBackend):
async def get(self, key: str):
value = await self.redis.get(key)
if value:
return pickle.loads(value)
return None
async def set(self, key: str, value: any, expire: int = None):
value = pickle.dumps(value)
await self.redis.set(key, value, expire)
@app.on_event("startup")
async def startup():
redis = aioredis.Redis(host='localhost', port=6379)
FastAPICache.init(CustomRedisBackend(redis), prefix="fastapi-cache:")
@app.get("/products/{product_id}")
@FastAPICache.cache(expire=300)
async def get_product(product_id: int):
return {"product_id": product_id}
后台任务
有效地管理长时间运行的操作至关重要。以下是处理后台任务的模式：
from fastapi import BackgroundTasks
from celery import Celery
celery_app = Celery('tasks', broker='redis://localhost:6379/0')
@celery_app.task
def process_video(video_id: int):
# Video processing logic
pass
@app.post("/videos/")
async def upload_video(
video: UploadFile,
background_tasks: BackgroundTasks
):
video_id = save_video(video)
background_tasks.add_task(process_video.delay, video_id)
return {"status": "processing"}
速率限制
保护 API 资源需要有效的速率限制。以下是基于 Redis 的实现：
from fastapi import HTTPException
import time
class RateLimiter:
def __init__(self, redis_client, limit: int, window: int):
self.redis = redis_client
self.limit = limit
self.window = window
async def is_allowed(self, key: str) -> bool:
current = int(time.time())
window_start = current - self.window
async with self.redis.pipeline() as pipe:
pipe.zremrangebyscore(key, 0, window_start)
pipe.zadd(key, {str(current): current})
pipe.zcard(key)
pipe.expire(key, self.window)
results = await pipe.execute()
return results[2] <= self.limit
@app.get("/api/resource")
async def get_resource(
redis: Redis = Depends(get_redis),
user: User = Depends(get_current_user)
):
rate_limiter = RateLimiter(redis, limit=100, window=3600)
if not await rate_limiter.is_allowed(f"rate_limit:{user.id}"):
raise HTTPException(status_code=429, detail="Rate limit exceeded")
return {"data": "resource"}
自定义中间件
中间件提供了强大的请求/响应修改功能：
from fastapi import Request
from fastapi.middleware.base import BaseHTTPMiddleware
import time
class TimingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
app.add_middleware(TimingMiddleware)
API 版本控制
维护 API 版本对于向后兼容性至关重要：
from fastapi import APIRouter
v1_router = APIRouter(prefix="/v1")
v2_router = APIRouter(prefix="/v2")
@v1_router.get("/users/{user_id}")
async def get_user_v1(user_id: int):
return {"version": "1", "user_id": user_id}
@v2_router.get("/users/{user_id}")
async def get_user_v2(user_id: int):
return {"version": "2", "user_id": user_id}
app.include_router(v1_router)
app.include_router(v2_router)
错误处理
一致的错误处理提高了 API 的可靠性：
from fastapi import HTTPException
from fastapi.responses import JSONResponse
class CustomException(Exception):
def __init__(self, message: str, code: str):
self.message = message
self.code = code
@app.exception_handler(CustomException)
async def custom_exception_handler(request, exc):
return JSONResponse(
status_code=400,
content={
"error": {
"code": exc.code,
"message": exc.message
}
}
)
@app.get("/items/{item_id}")
async def get_item(item_id: int):
if item_id < 0:
raise CustomException(
message="Invalid item ID",
code="INVALID_ID"
)
return {"item_id": item_id}
测试
全面的测试确保API的可靠性：
from fastapi.testclient import TestClient
import pytest
client = TestClient(app)
@pytest.fixture
def test_db():
# Setup test database
db = Database()
yield db
# Cleanup
def test_read_item():
response = client.get("/items/1")
assert response.status_code == 200
assert response.json() == {"item_id": 1}
def test_create_item():
response = client.post(
"/items/",
json={"name": "Test Item"}
)
assert response.status_code == 201
生产部署
对于生产部署，请考虑以下配置：
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
app = FastAPI(
title="Production API",
description="Production-ready FastAPI application",
version="1.0.0",
docs_url="/documentation",
redoc_url=None
)
app.add_middleware(
CORSMiddleware,
allow_origins=["https://allowed-domain.com"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
if __name__ == "__main__":
uvicorn.run(
"main:app",
host="0.0.0.0",
port=8000,
workers=4,
log_level="info",
reload=False
)
这些模式构成了强大的 FastAPI 应用程序的基础。异步功能与适当的资源管理和错误处理相结合，可创建高性能 API。定期测试和监控可确保生产环境中的可靠性。
请记住根据特定要求和规模调整这些模式。FastAPI 的灵活性允许自定义，同时保持性能和代码清晰度。