国产在线精品福利一区二区三区,一男一女一级毛片,羞羞视频免费观看入口

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()：

# 假裝這是個(gè)很慢的操作

return {“data”： “這數(shù)據(jù)算了老半天”}

溫馨提示：記得先把Redis服務(wù)跑起來(lái)，不然代碼會(huì)報(bào)錯(cuò)哦～

高級(jí)玩法來(lái)了

有時(shí)候我們想根據(jù)不同的請(qǐng)求參數(shù)來(lái)緩存，這也不是啥難事：

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



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



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



    # 這樣每個(gè)用戶的數(shù)據(jù)都會(huì)分別緩存



    return {“user_id”： user_id， “l(fā)anguage”： lang}

有個(gè)小坑要提醒一下，如果你的接口返回的是模型對(duì)象，得先把它轉(zhuǎn)成dict，不然緩存會(huì)出問(wèn)題：

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()  # 這里要轉(zhuǎn)換一下

動(dòng)態(tài)設(shè)置緩存時(shí)間

有時(shí)候我想動(dòng)態(tài)決定緩存多久，可以這么搞：

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”： “我只緩存半分鐘”}

手動(dòng)管理緩存

緩存有時(shí)候得手動(dòng)清理，比如數(shù)據(jù)更新了：

from fastapi_cache import FastAPICache



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



async def update_data()：



    # 清除指定的緩存



    await FastAPICache.clear(namespace=“user_profile”)



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

最后說(shuō)兩句，這個(gè)庫(kù)真的挺好使的，但也別啥都往緩存里塞。實(shí)時(shí)性要求高的數(shù)據(jù)就別緩存了，不然數(shù)據(jù)不準(zhǔn)還得背鍋。緩存時(shí)間也得掌握好，太長(zhǎng)容易數(shù)據(jù)不同步，太短又起不到緩存的作用。

你要是想讓接口速度飛起來(lái)，不妨試試這個(gè)庫(kù)，保證讓你爽到飛起～# FastAPI-Cache2進(jìn)階：3個(gè)實(shí)用技巧助你玩轉(zhuǎn)緩存

平時(shí)寫(xiě)接口時(shí)，咱不能光會(huì)用最基本的緩存功能。今天接著上次的內(nèi)容，再聊聊FastAPI-Cache2的幾個(gè)進(jìn)階用法，這些都是我踩過(guò)坑總結(jié)出來(lái)的干貨。

自定義緩存KEY

默認(rèn)情況下，FastAPI-Cache2用請(qǐng)求路徑和參數(shù)生成緩存key。但有時(shí)候這樣不夠靈活，看看怎么自定義：

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}的數(shù)據(jù)”}

溫馨提示：自定義key的時(shí)候要當(dāng)心，key太長(zhǎng)會(huì)影響Redis性能，太短又容易撞車(chē)。找到平衡點(diǎn)很重要！

條件性緩存

有些情況下，我們希望根據(jù)具體條件決定要不要使用緩存：

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}用戶的數(shù)據(jù)”}

緩存預(yù)熱

系統(tǒng)剛啟動(dòng)時(shí)，緩存是空的，第一波請(qǐng)求會(huì)比較慢。咱可以提前把常用數(shù)據(jù)加載到緩存里：

from fastapi_cache import FastAPICache



import asyncio



async def warm_up_cache()：



    # 預(yù)熱一些常用數(shù)據(jù)



    data = await get_common_data()  # 假設(shè)這是個(gè)獲取數(shù)據(jù)的函數(shù)



    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”)



    # 異步預(yù)熱緩存



    asyncio.create_task(warm_up_cache())



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



async def get_cached_common_data()：



    # 優(yōu)先從緩存拿數(shù)據(jù)



    data = await FastAPICache.get(“common_data”)



    if not data：



        # 緩存沒(méi)有再去查詢



        data = await get_common_data()



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



    return data