FastAPI-Cache2：一个让接口飞起来的缓存神器

写过接口的小伙伴都知道，有些数据库查询特别费时，还有些计算逻辑贼复杂。每次请求都去算一遍，那接口响应速度慢得像蜗牛。这不，我发现了个宝贝 – FastAPI-Cache2 ，直接把接口响应速度提升好几倍，今天就跟大家唠唠它的使用秘诀。

安装配置两步搞定

先把这玩意装上：

pip install fastapi-cache2[redis]

我一般用Redis当存储后端，简单粗暴。当然你也能用别的，比如内存缓存啥的。

基础用法就是这么简单

来看看最基本的用法：

from fastapi import FastAPI



from fastapi_cache import FastAPICache



from fastapi_cache.backends.redis import RedisBackend



from fastapi_cache.decorator import cache



from redis import asyncio as aioredis



app = FastAPI()



@app.on_event(“startup”)



async def startup()：



    redis = aioredis.from_url(“redis：//localhost”)



    FastAPICache.init(RedisBackend(redis)， prefix=“myapp-cache”)



@app.get(“/slow-data”)



@cache(expire=60)  # 缓存一分钟



async def get_slow_data()：



    # 假装这是个很慢的操作



    return {“data”： “这数据算了老半天”}

温馨提示：记得先把Redis服务跑起来，不然代码会报错哦～

高级玩法来了

有时候我们想根据不同的请求参数来缓存，这也不是啥难事：

@app.get(“/user/{user_id}”)



@cache(expire=60， namespace=“user_profile”)



async def get_user(user_id： int， lang： str = “zh”)：



    # 这样每个用户的数据都会分别缓存



    return {“user_id”： user_id， “language”： lang}

有个小坑要提醒一下，如果你的接口返回的是模型对象，得先把它转成dict，不然缓存会出问题：

from pydantic import BaseModel



class User(BaseModel)：



    name： str



    age： int



@app.get(“/user-model/{user_id}”)



@cache(expire=60)



async def get_user_model(user_id： int)：



    user = User(name=“凡凡”， age=18)



    return user.dict()  # 这里要转换一下

动态设置缓存时间

有时候我想动态决定缓存多久，可以这么搞：

def dynamic_expire(ttl： int = 60)：



    def wrapper(func)：



        @cache(expire=ttl)



        async def inner(*args， **kwargs)：



            return await func(*args， **kwargs)



        return inner



    return wrapper



@app.get(“/dynamic-cache”)



@dynamic_expire(ttl=30)  # 这里可以随意改



async def get_dynamic_data()：



    return {“msg”： “我只缓存半分钟”}

手动管理缓存

缓存有时候得手动清理，比如数据更新了：

from fastapi_cache import FastAPICache



@app.post(“/update-data”)



async def update_data()：



    # 清除指定的缓存



    await FastAPICache.clear(namespace=“user_profile”)



    return {“msg”： “缓存已清除”}

最后说两句，这个库真的挺好使的，但也别啥都往缓存里塞。实时性要求高的数据就别缓存了，不然数据不准还得背锅。缓存时间也得掌握好，太长容易数据不同步，太短又起不到缓存的作用。

你要是想让接口速度飞起来，不妨试试这个库，保证让你爽到飞起～# FastAPI-Cache2进阶：3个实用技巧助你玩转缓存

平时写接口时，咱不能光会用最基本的缓存功能。今天接着上次的内容，再聊聊FastAPI-Cache2的几个进阶用法，这些都是我踩过坑总结出来的干货。

自定义缓存KEY

默认情况下，FastAPI-Cache2用请求路径和参数生成缓存key。但有时候这样不够灵活，看看怎么自定义：

from fastapi_cache.decorator import cache



from fastapi import Query



def my_key_builder(



    func，



    namespace： str = “”，



    user_id： int = None，



    query_params： dict = None，



)：



    # 自己组装缓存key



    prefix = f“{namespace}：{func.__module__}：{func.__name__}：”



    dynamic_key = f“user_{user_id}”



    if query_params：



        dynamic_key += f“_{sorted(query_params.items())}”



    return prefix + dynamic_key



@app.get(“/custom-cache”)



@cache(expire=60， key_builder=my_key_builder)



async def get_data(user_id： int = Query(None))：



    return {“data”： f“用户{user_id}的数据”}

温馨提示：自定义key的时候要当心，key太长会影响Redis性能，太短又容易撞车。找到平衡点很重要！

条件性缓存

有些情况下，我们希望根据具体条件决定要不要使用缓存：

from typing import Optional



from functools import wraps



def smart_cache(expire： int = 60， condition： callable = None)：



    def wrapper(func)：



        @wraps(func)



        async def inner(*args， **kwargs)：



            # 判断是否需要缓存



            if condition and not condition(*args， **kwargs)：



                return await func(*args， **kwargs)



            # 需要缓存的情况



            @cache(expire=expire)



            async def cached_func(*a， **kw)：



                return await func(*a， **kw)



            return await cached_func(*args， **kwargs)



        return inner



    return wrapper



def should_cache(user_type： str) -> bool：



    return user_type == “normal”



@app.get(“/smart-cache/{user_type}”)



@smart_cache(expire=30， condition=should_cache)



async def get_user_data(user_type： str)：



    # VIP用户不走缓存，普通用户走缓存



    return {“msg”： f“{user_type}用户的数据”}

缓存预热

系统刚启动时，缓存是空的，第一波请求会比较慢。咱可以提前把常用数据加载到缓存里：

from fastapi_cache import FastAPICache



import asyncio



async def warm_up_cache()：



    # 预热一些常用数据



    data = await get_common_data()  # 假设这是个获取数据的函数



    key = “common_data”



    await FastAPICache.set(key， data， expire=3600)



@app.on_event(“startup”)



async def startup()：



    # Redis连接配置



    redis = aioredis.from_url(“redis：//localhost”)



    FastAPICache.init(RedisBackend(redis)， prefix=“myapp-cache”)



    # 异步预热缓存



    asyncio.create_task(warm_up_cache())



@app.get(“/common-data”)



async def get_cached_common_data()：



    # 优先从缓存拿数据



    data = await FastAPICache.get(“common_data”)



    if not data：



        # 缓存没有再去查询



        data = await get_common_data()



        await FastAPICache.set(“common_data”， data， expire=3600)



    return data