日产精品1区二区三区乱码,精品三级66在线播放,久精品在线观看

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