1. 🎯 FastAPI 为何脱颖而出？
FastAPI 的三大杀手锏：
- 基于 Python 类型提示的自动数据验证
- 原生异步支持，性能媲美 NodeJS
- 自动生成交互式 API 文档
2. 🏗️ 架构设计：异步优先
2.1 真正的异步支持
FastAPI 从底层为异步设计，不是简单包装。对比测试数据：
| 场景 | FastAPI | Flask+gevent | 性能提升 |
|---|---|---|---|
| 1000 并发 IO 操作 | 1.8 秒 | 2.9 秒 | 61% |
| 混合负载 | 2.1 秒 | 3.4 秒 | 62% |
| 内存占用 | 85MB | 120MB | 29% |
# 真正的异步处理 from fastapi import FastAPI, BackgroundTasks import asyncio from datetime import datetime app = FastAPI() @app.get("/api/status") async def get_status(): """完全异步的端点""" start = datetime.now() # 并发执行多个 IO 操作 results = await asyncio.gather( fetch_user_data(), fetch_order_data(), fetch_product_data() ) return { "time": (datetime.now() - start).total_seconds(), "data": results } async def fetch_user_data(): await asyncio.sleep(0.1) # 模拟数据库查询 return {"users": 150} async def fetch_order_data(): await asyncio.sleep(0.2) return {"orders": 45} async def fetch_product_data(): await asyncio.sleep(0.15) return {"products": 89}
3. 🔧 依赖注入：FastAPI 的灵魂
3.1 依赖注入三层架构
3.2 实战：企业级依赖系统
from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from typing import Optional, Generator from contextlib import contextmanager import redis from datetime import datetime, timedelta # 1. 认证依赖 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token") async def get_current_user(token: str = Depends(oauth2_scheme)): """用户认证依赖""" try: # 解码 JWT 令牌 payload = decode_jwt(token) return { "id": payload["user_id"], "username": payload["sub"], "roles": payload.get("roles", []) } except Exception: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的认证令牌" ) # 2. 角色权限依赖 def require_roles(required_roles: list): """角色检查依赖工厂""" async def check_roles( current_user: dict = Depends(get_current_user) ): user_roles = set(current_user.get("roles", [])) required = set(required_roles) if not required.intersection(user_roles): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="权限不足" ) return current_user return check_roles # 3. 数据库会话依赖 class DatabaseSession: """数据库会话管理""" def __init__(self, readonly: bool = False): self.readonly = readonly async def __call__(self): # 创建数据库会话 session = create_db_session(readonly=self.readonly) try: yield session finally: session.close() # 4. Redis 缓存依赖 @contextmanager def get_redis_connection(): """Redis 连接管理""" client = redis.Redis.from_url("redis://localhost:6379") try: yield client finally: client.close() # 5. 在路由中使用 @app.get("/admin/users") async def get_users( db = Depends(DatabaseSession()), cache = Depends(get_redis_connection), admin_only: bool = Depends(require_roles(["admin"])) ): """管理员获取用户列表""" # 检查缓存 cached = cache.get("all_users") if cached: return json.loads(cached) # 查询数据库 users = db.query(User).all() result = [user.dict() for user in users] # 设置缓存 cache.setex("all_users", 300, json.dumps(result)) return result
