请求流程（无缓存）:
客户端 → API → 数据库 → 返回结果
(每次都要查数据库)
请求流程（有缓存）:
客户端 → API → 缓存命中 → 返回结果
(90%请求直接返回)

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import time
from functools import wraps
import hashlib
app = FastAPI()
简单内存缓存
cache = {}
CACHE_TTL = 300  # 5分钟
def get_cache_key(path: str, query_params: dict) -> str:
"""生成缓存键"""
key_string = f"{path}:{str(query_params)}"
return hashlib.md5(key_string.encode()).hexdigest()
def is_cache_expired(timestamp: float) -> bool:
"""检查缓存是否过期"""
return time.time() - timestamp > CACHE_TTL
@app.middleware("http")
async def cache_middleware(request: Request, call_next):
"""缓存中间件"""
# 只缓存GET请求
if request.method != "GET":
return await call_next(request)
# 生成缓存键
cache_key = get_cache_key(
request.url.path,
dict(request.query_params)
)
# 检查缓存
if cache_key in cache:
cached_data, timestamp = cache[cache_key]
if not is_cache_expired(timestamp):
return JSONResponse(cached_data)
# 执行请求
response = await call_next(request)
# 缓存响应（只缓存200状态码）
if response.status_code == 200:
body = await response.body()
cache[cache_key] = (
{"status": response.status_code, "body": body.decode()},
time.time()
)
return response

pip install redis

import redis
import json
from datetime import timedelta
连接Redis
redis_client = redis.Redis(
host="localhost",
port=6379,
db=0,
decode_responses=True
)
连接池（推荐）
pool = redis.ConnectionPool(
host="localhost",
port=6379,
db=0,
max_connections=20,
decode_responses=True
)
redis_client = redis.Redis(connection_pool=pool)

from typing import Any, Optional
class Cache:
"""缓存工具类"""
@staticmethod
def get(key: str) -> Optional[Any]:
"""获取缓存"""
value = redis_client.get(key)
if value is None:
return None
try:
return json.loads(value)
except json.JSONDecodeError:
return value
@staticmethod
def set(key: str, value: Any, expire: int = 300) -> bool:
"""设置缓存"""
try:
redis_client.setex(
key,
timedelta(seconds=expire),
json.dumps(value)
)
return True
except Exception as e:
print(f"缓存设置失败: {e}")
return False
@staticmethod
def delete(key: str) -> bool:
"""删除缓存"""
return bool(redis_client.delete(key))
@staticmethod
def exists(key: str) -> bool:
"""检查缓存是否存在"""
return bool(redis_client.exists(key))
@staticmethod
def increment(key: str, amount: int = 1) -> int:
"""递增"""
return redis_client.incr(key, amount)
@staticmethod
def expire(key: str, seconds: int) -> bool:
"""设置过期时间"""
return bool(redis_client.expire(key, seconds))
使用
Cache.set("user:1", {"id": 1, "name": "Alice"}, expire=3600)
user = Cache.get("user:1")
Cache.delete("user:1")

from fastapi import FastAPI, Depends
from pydantic import BaseModel
from typing import Optional
app = FastAPI()
class User(BaseModel):
id: int
name: str
email: str
模拟数据库
users_db = {
1: User(id=1, name="Alice", email="alice@example.com"),
2: User(id=2, name="Bob", email="bob@example.com"),
}
def get_user_from_db(user_id: int) -> Optional[User]:
"""从数据库获取用户"""
return users_db.get(user_id)
@app.get("/users/{user_id}", response_model=User)
async def get_user(user_id: int):
"""获取用户（带缓存）"""
cache_key = f"user:{user_id}"
# 尝试从缓存获取
cached_user = Cache.get(cache_key)
if cached_user:
return cached_user
# 从数据库获取
user = get_user_from_db(user_id)
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
# 写入缓存
Cache.set(cache_key, user.model_dump(), expire=300)
return user

# 问题：查询不存在的数据，缓存和数据库都没有
解决：缓存空值
@app.get("/users/{user_id}", response_model=User)
async def get_user_safe(user_id: int):
"""安全获取用户（防穿透）"""
cache_key = f"user:{user_id}"
# 尝试从缓存获取
cached_user = Cache.get(cache_key)
# 空值标记（防止穿透）
if cached_user == "__NOT_FOUND__":
raise HTTPException(status_code=404, detail="用户不存在")
if cached_user:
return cached_user
# 从数据库获取
user = get_user_from_db(user_id)
if not user:
# 缓存空值（设置短过期时间）
Cache.set(cache_key, "__NOT_FOUND__", expire=60)
raise HTTPException(status_code=404, detail="用户不存在")
# 写入缓存
Cache.set(cache_key, user.model_dump(), expire=300)
return user

import asyncio
from functools import wraps
async_locks = {}
async def get_user_with_lock(user_id: int) -> User:
"""带锁获取用户（防击穿）"""
cache_key = f"user:{user_id}"
# 尝试从缓存获取
cached_user = Cache.get(cache_key)
if cached_user and cached_user != "__NOT_FOUND__":
return cached_user
# 获取锁
lock_key = f"lock:user:{user_id}"
# 尝试获取锁
acquired = redis_client.set(lock_key, "1", nx=True, ex=10)
if not acquired:
# 等待锁释放后重试
await asyncio.sleep(0.1)
return await get_user_with_lock(user_id)
try:
# 双重检查缓存
cached_user = Cache.get(cache_key)
if cached_user and cached_user != "__NOT_FOUND__":
return cached_user
# 从数据库获取
user = get_user_from_db(user_id)
if not user:
Cache.set(cache_key, "__NOT_FOUND__", expire=60)
raise HTTPException(status_code=404, detail="用户不存在")
# 写入缓存
Cache.set(cache_key, user.model_dump(), expire=300)
return user
finally:
# 释放锁
redis_client.delete(lock_key)

import random
def get_user_with_random_expire(user_id: int) -> User:
"""带随机过期时间（防雪崩）"""
cache_key = f"user:{user_id}"
# 随机过期时间（基础时间 ± 随机偏移）
base_expire = 300
random_offset = random.randint(-60, 60)
expire = max(60, base_expire + random_offset)
cached_user = Cache.get(cache_key)
if cached_user and cached_user != "__NOT_FOUND__":
return cached_user
user = get_user_from_db(user_id)
if not user:
Cache.set(cache_key, "__NOT_FOUND__", expire=60)
raise HTTPException(status_code=404, detail="用户不存在")
# 写入缓存（随机过期时间）
Cache.set(cache_key, user.model_dump(), expire=expire)
return user

from functools import wraps
import hashlib
import inspect
def cache_result(expire: int = 300, prefix: str = "func"):
"""函数结果缓存装饰器"""
def decorator(func):
@wraps(func)
async def wrapper(args, *kwargs):
# 生成缓存键
sig = inspect.signature(func)
bound = sig.bind(args, *kwargs)
bound.apply_defaults()
key_parts = [prefix, func.__name__]
for name, value in bound.arguments.items():
if isinstance(value, (str, int, float, bool)):
key_parts.append(f"{name}={value}")
cache_key = ":".join(key_parts)
# 尝试从缓存获取
cached = Cache.get(cache_key)
if cached is not None:
return cached
# 执行函数
result = await func(args, *kwargs)
# 写入缓存
Cache.set(cache_key, result, expire=expire)
return result
return wrapper
return decorator
使用
@cache_result(expire=600, prefix="user")
async def get_user_data(user_id: int):
"""获取用户数据（带缓存）"""
# 模拟耗时操作
await asyncio.sleep(1)
return {"id": user_id, "name": "Alice"}
@app.get("/users/{user_id}/data")
async def get_user_data_endpoint(user_id: int):
return await get_user_data(user_id)

def cache_method(expire: int = 300):
"""类方法缓存装饰器"""
def decorator(func):
@wraps(func)
async def wrapper(self, args, *kwargs):
cache_key = f"{self.__class__.__name__}:{func.__name__}:{args}:{kwargs}"
cached = Cache.get(cache_key)
if cached is not None:
return cached
result = await func(self, args, *kwargs)
Cache.set(cache_key, result, expire=expire)
return result
return wrapper
return decorator
class UserService:
@cache_method(expire=300)
async def get_user_profile(self, user_id: int):
return {"id": user_id, "profile": "data"}

from fastapi import Response
from datetime import datetime, timedelta
@app.get("/static/{filename}")
async def serve_static(filename: str, response: Response):
"""静态文件（带HTTP缓存）"""
file_path = Path("static") / filename
if not file_path.exists():
raise HTTPException(status_code=404, detail="文件不存在")
# 设置缓存头
response.headers["Cache-Control"] = "public, max-age=31536000"  # 1年
response.headers["ETag"] = f'"{file_path.stat().st_mtime}"'
return FileResponse(file_path)

@app.get("/api/data")
async def get_data(request: Request):
"""支持条件请求"""
# 生成ETag
data = {"version": "1.0", "items": [...]}
etag = hashlib.md5(json.dumps(data).encode()).hexdigest()
# 检查If-None-Match
if_none_match = request.headers.get("If-None-Match")
if if_none_match and if_none_match.strip('"') == etag:
return Response(status_code=304)
response = JSONResponse(data)
response.headers["ETag"] = f'"{etag}"'
response.headers["Cache-Control"] = "public, max-age=60"
return response

@app.get("/api/config")
async def get_config(request: Request):
"""支持Last-Modified"""
config = {"setting": "value"}
# 获取最后修改时间
last_modified = "Wed, 26 May 2026 12:00:00 GMT"
# 检查If-Modified-Since
if_modified_since = request.headers.get("If-Modified-Since")
if if_modified_since == last_modified:
return Response(status_code=304)
response = JSONResponse(config)
response.headers["Last-Modified"] = last_modified
response.headers["Cache-Control"] = "public, max-age=60"
return response

from sqlalchemy.orm import Session
from typing import Optional, List
def get_user_with_cache(db: Session, user_id: int) -> Optional[User]:
"""带缓存的用户查询"""
cache_key = f"db:user:{user_id}"
# 尝试从缓存获取
cached = Cache.get(cache_key)
if cached:
return User(**cached)
# 从数据库查询
user = db.query(UserModel).filter(UserModel.id == user_id).first()
if not user:
return None
# 写入缓存
Cache.set(cache_key, user.model_dump(), expire=300)
return user
def get_users_with_cache(db: Session, skip: int = 0, limit: int = 100) -> List[User]:
"""带缓存的用户列表查询"""
cache_key = f"db:users:{skip}:{limit}"
cached = Cache.get(cache_key)
if cached:
return [User(**u) for u in cached]
users = db.query(UserModel).offset(skip).limit(limit).all()
if not users:
return []
Cache.set(cache_key, [u.model_dump() for u in users], expire=300)
return [User.model_validate(u) for u in users]

def invalidate_user_cache(user_id: int):
"""失效用户缓存"""
Cache.delete(f"user:{user_id}")
Cache.delete(f"db:user:{user_id}")
def invalidate_users_list_cache():
"""失效用户列表缓存"""
# 删除所有用户列表缓存
keys = redis_client.keys("db:users:*")
if keys:
redis_client.delete(*keys)
@app.put("/users/{user_id}")
async def update_user(user_id: int, user_in: UserUpdate, db: Session = Depends(get_db)):
"""更新用户（同时失效缓存）"""
user = get_user_with_cache(db, user_id)
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
# 更新数据库
for field, value in user_in.model_dump(exclude_unset=True).items():
setattr(user, field, value)
db.commit()
# 失效缓存
invalidate_user_cache(user_id)
invalidate_users_list_cache()
return user

from collections import OrderedDict
import threading
class LRUCache:
"""LRU内存缓存"""
def __init__(self, max_size: int = 1000):
self.max_size = max_size
self.cache = OrderedDict()
self.lock = threading.Lock()
def get(self, key: str) -> Optional[Any]:
with self.lock:
if key not in self.cache:
return None
# 移到末尾（最近使用）
self.cache.move_to_end(key)
return self.cache[key]
def set(self, key: str, value: Any):
with self.lock:
if key in self.cache:
self.cache.move_to_end(key)
else:
if len(self.cache) >= self.max_size:
# 移除最久未使用
self.cache.popitem(last=False)
self.cache[key] = value
def delete(self, key: str):
with self.lock:
self.cache.pop(key, None)
L1: 内存缓存（快速）
l1_cache = LRUCache(max_size=1000)
L2: Redis缓存（持久）
l2_cache = Cache
class MultiLevelCache:
"""多级缓存"""
@staticmethod
def get(key: str) -> Optional[Any]:
# L1
value = l1_cache.get(key)
if value is not None:
return value
# L2
value = l2_cache.get(key)
if value is not None:
# 回写L1
l1_cache.set(key, value)
return value
return None
@staticmethod
def set(key: str, value: Any, expire: int = 300):
# L1
l1_cache.set(key, value)
# L2
l2_cache.set(key, value, expire=expire)
@staticmethod
def delete(key: str):
l1_cache.delete(key)
l2_cache.delete(key)
使用
MultiLevelCache.set("user:1", {"id": 1, "name": "Alice"})
user = MultiLevelCache.get("user:1")

from fastapi import FastAPI, Depends, HTTPException
from fastapi.responses import Response
from pydantic import BaseModel
from sqlalchemy.orm import Session
from datetime import datetime
import hashlib
import json
app = FastAPI()
class User(BaseModel):
id: int
name: str
email: str
updated_at: datetime
class CacheManager:
"""缓存管理器"""
def __init__(self):
self.l1_cache = LRUCache(max_size=500)
self.l2_client = redis.Redis(host="localhost", decode_responses=True)
def _get_l2(self, key: str) -> Optional[dict]:
value = self.l2_client.get(key)
if value:
return json.loads(value)
return None
def _set_l2(self, key: str, value: dict, expire: int = 300):
self.l2_client.setex(key, expire, json.dumps(value))
def get(self, key: str) -> Optional[dict]:
# L1
value = self.l1_cache.get(key)
if value is not None:
return value
# L2
value = self._get_l2(key)
if value is not None:
self.l1_cache.set(key, value)
return value
return None
def set(self, key: str, value: dict, expire: int = 300):
self.l1_cache.set(key, value)
self._set_l2(key, value, expire)
def delete(self, key: str):
self.l1_cache.delete(key)
self.l2_client.delete(key)
def delete_pattern(self, pattern: str):
keys = self.l2_client.keys(pattern)
if keys:
self.l2_client.delete(*keys)
cache_manager = CacheManager()
@app.get("/api/users/{user_id}", response_model=User)
async def get_user(
user_id: int,
db: Session = Depends(get_db)
):
"""获取用户（多级缓存）"""
cache_key = f"user:{user_id}"
# 尝试从缓存获取
cached = cache_manager.get(cache_key)
if cached:
return User(**cached)
# 从数据库获取
user = db.query(UserModel).filter(UserModel.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
# 写入缓存
cache_manager.set(cache_key, user.model_dump(), expire=300)
return user
@app.get("/api/users", response_model=list[User])
async def get_users(
skip: int = 0,
limit: int = 20,
db: Session = Depends(get_db)
):
"""获取用户列表（带缓存）"""
cache_key = f"users:list:{skip}:{limit}"
cached = cache_manager.get(cache_key)
if cached:
return [User(**u) for u in cached]
users = db.query(UserModel).offset(skip).limit(limit).all()
if not users:
return []
cache_manager.set(cache_key, [u.model_dump() for u in users], expire=300)
return [User.model_validate(u) for u in users]
@app.post("/api/users")
async def create_user(user_in: UserCreate, db: Session = Depends(get_db)):
"""创建用户（失效缓存）"""
user = UserCreate(**user_in.model_dump())
db.add(user)
db.commit()
db.refresh(user)
# 失效缓存
cache_manager.delete_pattern("users:*")
return user
@app.put("/api/users/{user_id}")
async def update_user(
user_id: int,
user_in: UserUpdate,
db: Session = Depends(get_db)
):
"""更新用户（失效缓存）"""
user = db.query(UserModel).filter(UserModel.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
for field, value in user_in.model_dump(exclude_unset=True).items():
setattr(user, field, value)
db.commit()
# 失效缓存
cache_manager.delete(f"user:{user_id}")
cache_manager.delete_pattern("users:list:*")
return user
@app.delete("/api/users/{user_id}")
async def delete_user(user_id: int, db: Session = Depends(get_db)):
"""删除用户（失效缓存）"""
user = db.query(UserModel).filter(UserModel.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
db.delete(user)
db.commit()
# 失效缓存
cache_manager.delete(f"user:{user_id}")
cache_manager.delete_pattern("users:list:*")
return {"message": "删除成功"}

📚 推荐阅读

KEYWORDS

IL, Python, pip, 函数, 列表

💡 如果你觉得这篇文章有帮助，请点个在看，分享给更多需要的人！

📝 关注我，获取更多实用干货～

🤝 有问题欢迎评论区留言交流！