聊聊我在 FastAPI 项目里踩的一个缓存坑

起因

事情是这样的。

我这个后端模板用了一套自定义的响应类，Success、Fail、SuccessExtra，都继承自 FastAPI 的 JSONResponse。当时的想法很简单——统一响应格式，前端拿到的数据结构永远是 {code, msg, data} 这个样子，分页接口再多带几个字段。

class Success(JSONResponse):
    def __init__(self, code: int = 200, msg: str = "OK", data: Any = None, **kwargs):
        content = {"code": code, "msg": msg, "data": data}
        super().__init__(content=content, status_code=code)

用起来很爽，Service 层直接 return Success(data=user_info) 就完事了，不用在 API 层再包一层。

问题出在我想给接口加缓存的时候。

第一版：天真的实现

最开始我写的缓存装饰器很简单：

def cached(prefix: str, ttl: int = 300):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            cache_key = f"{prefix}:{args}:{kwargs}"

            # 从 Redis 取
            cached_result = await redis.get(cache_key)
            if cached_result:
                return json.loads(cached_result)

            # 执行函数
            result = await func(*args, **kwargs)

            # 存 Redis
            await redis.setex(cache_key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

然后给用户详情接口加上：

@cached("user_detail")
async def get_user_detail(self, user_id: int) -> Success:
    user = await self.repository.get(user_id)
    return Success(data=user.to_dict())

跑起来，第一次请求正常，第二次请求——炸了。

1	TypeError: Object of type Success is not JSON serializable

我愣了一下，反应过来了：Success 是个 JSONResponse 对象，不是普通的 dict，json.dumps 序列化不了。

第二版：绕过去

最直观的解法是：别缓存 Response 对象，缓存原始数据。

@cached("user_detail")
async def get_user_data(self, user_id: int) -> dict:
    user = await self.repository.get(user_id)
    return user.to_dict()  # 返回 dict

async def get_user_detail(self, user_id: int) -> Success:
    data = await self.get_user_data(user_id)
    return Success(data=data)

能跑。但是丑。

我得把每个需要缓存的方法拆成两个：一个返回原始数据用来缓存，一个包装成 Response 给上层用。代码量翻倍，而且容易忘记——哪天改了 get_user_data 忘了同步改 get_user_detail，排查起来又是一通。

更烦的是分页接口。SuccessExtra 除了 data 还有 total、page、page_size，拆出来的数据结构越来越复杂，维护成本蹭蹭涨。

我不想这么搞。

第三版：把 Response 拆开存

既然问题是 JSONResponse 不能直接序列化，那我就手动拆开。

JSONResponse 有个 body 属性，存的是已经 JSON 序列化后的字节流。我可以把这个 body 拿出来，存进 Redis；取出来的时候，再用这个 body 重建一个 Response 对象。

但是光存 body 不够，我还得知道这是 Success 还是 Fail 还是 SuccessExtra，不然没法重建。

于是最终的缓存结构长这样：

{
    "__response__": True,           # 标记这是个响应对象
    "class": "Success",             # 响应类型
    "payload": {"code": 200, ...}   # 响应体内容
}

存的时候：

if isinstance(result, (Success, Fail, SuccessExtra)):
    # 从 body 里反序列化出原始内容
    body_bytes = result.body
    payload = json.loads(body_bytes.decode("utf-8"))

    value_to_cache = {
        "__response__": True,
        "class": result.__class__.__name__,
        "payload": payload,
    }
    await cache_manager.set(cache_key, value_to_cache, ttl)

取的时候：

cached_result = await cache_manager.get(cache_key)
if cached_result and cached_result.get("__response__"):
    response_type = cached_result.get("class")
    payload = cached_result.get("payload", {})

    response_cls = {
        "Success": Success,
        "Fail": Fail,
        "SuccessExtra": SuccessExtra,
    }.get(response_type, Success)

    return response_cls(**payload)

核心思路就是：存的时候解构，取的时候重建。

这么做值得吗

说实话，这个实现有点绕。有朋友看了代码问我，为什么不直接让 Service 层返回普通数据，在 API 层统一包装响应？

这个问题我也想过。两种方式各有利弊：

Service 返回 Response 对象的好处：

业务逻辑可以直接决定响应码和消息，比如校验失败时 return Fail(msg="用户名已存在")
API 层更薄，基本就是个路由转发
异常情况不用抛 Exception，代码流更直观

Service 返回原始数据的好处：

缓存简单，不用处理 Response 对象
Service 更纯粹，不依赖 HTTP 层的概念
复用性更好，比如后面加个 gRPC 接口也能用

我最后选了前者，主要是因为这个模板面向的场景就是 HTTP API，不太考虑协议切换。而且业务逻辑里经常需要根据不同情况返回不同的 code 和 msg，放在 Service 层处理更自然。

至于缓存的复杂度——反正就写一次，封装好了上层用起来没区别。

还有一个坑：缓存键的生成

缓存装饰器还有个问题是键怎么生成。

最开始我直接把 args 和 kwargs 拼进去，结果发现 Service 方法的第一个参数是 self，它的字符串表示长这样：

1	<services.user.UserService object at 0x7f8b8c0d5a90>

每次请求这个地址都不一样，缓存永远命中不了。

后来改成跳过 self，只用业务参数生成键：

def cache_key(self, prefix: str, *args, **kwargs) -> str:
    key_parts = [prefix]

    # args[0] 是 self，跳过
    if len(args) > 1:
        key_parts.extend(str(arg) for arg in args[1:])

    if kwargs:
        sorted_kwargs = sorted(kwargs.items())
        key_parts.extend(f"{k}:{v}" for k, v in sorted_kwargs)

    return ":".join(key_parts)

这样 get_user_detail(user_id=123) 的缓存键就是 user_detail:123。

但后来又发现，有些接口需要更复杂的键生成逻辑，比如要把当前用户 ID 也算进去（不同用户看到的数据可能不同）。于是加了个 key_func 参数，允许传入自定义的键生成函数：

def cached(prefix: str, ttl: int = None, key_func: Callable = None):
    def decorator(func):
        async def wrapper(*args, **kwargs):
            if key_func:
                cache_key = key_func(*args, **kwargs)
            else:
                cache_key = cache_manager.cache_key(prefix, *args, **kwargs)
            # ...

最后

这套缓存方案跑了挺久，没出过什么问题。

回头看，其实核心就一个思路：碰到不能直接序列化的对象，就想办法把它拆成能序列化的结构，再存一个类型标记用于重建。这个模式挺通用的，不只是 Response 对象，很多自定义类型都可以这么处理。

当然，如果你的项目架构允许 Service 层返回纯数据，那直接返回 dict 或 Pydantic model 肯定是更简单的选择。我这么搞纯粹是因为习惯了这种写法，懒得改。

代码写多了就会发现，很多时候技术选型没有绝对的对错，只有适不适合。能解决问题、好维护、同事看得懂，就是好代码。