python之fastAPI实战篇(完结版)

今天讲一下缓存

1.缓存

缓存就是把数据存入redis里面，查询时候先通过redis查，redis没有的话通过数据库查

流程

1.安装redis客户端

pip install redis

2.引用redis

import redis.asyncio as redisredis_client=redis.Redis("localhost",6379,0,decode_responses=True)

3.封装读取和写入操作

async def get_cache(key:str):    try:return await redis_client.get(key)    except Exception as e:print("获取缓存失败")return Noneasync def get_json_cache(key: str):    try:        data= await redis_client.get(key)if data:return json.loads(data)return None    except Exception as e:print("获取缓存失败")return Noneasync def set_cache(key: str, value: Any, expire: int = 3600):    try:if isinstance(value,(dict,list)):            value=json.dumps(value,ensure_ascii=False)        await redis_client.setex(key, expire, value)return True    except Exception as e:return False

示例：将分类信息存入缓存

@router.get("/categories")async def getcategory(db:AsyncSession=Depends(getdatabase)):    data=await get_json_cache("news:category:")print(data,"1233")if data == None:        result = await db.execute(select(NewsCategory))        data = result.scalars().all()print(data,"data")        await set_cache("news:category:",data,7200)return {"msg":"获取成功","data":data,"code":200}def news_category_to_dict(obj):if isinstance(obj, NewsCategory):return obj.__dict__async def set_cache(key: str, value: Any, expire: int = 3600):    try:if isinstance(value,(dict,list)):            value=json.dumps(value,default=news_category_to_dict)        await redis_client.setex(key,expire,value )return True    except Exception as e:print(f"存入失败{e}")return False