- 在配置redis的时候遇到了一个神奇的问题
在同一个请求中 使用redis可以set和get内容
另一个请求中却不能get到之前set的内容
1. 注册redis到app
参考了别人的项目和redis的官方文档, 在app启动的时候增加了redis的注册
from redis import asyncio
@app.on_event("startup")
async def startup_event():
app.redis = await asyncio.from_url(REDIS_URL, decode_responses=True, encoding="utf8", )
然后在后续的项目中通过request.app.redis.get/set 来插入和获得内容
2. 增加通用工具方法
在工具类中增加了set_cache 和 get_cache del_cache
import json
from datetime import timedelta
from fastapi import Request
async def set_cache(key: str, value, request: Request, ex: timedelta = None) -> bool:
if not all((key, value)):
return False
params = {'name': 'blog.cache.' + key, 'value': value}
if ex:
params['ex'] = ex
if isinstance(value, int | float | str):
return await request.app.redis.set(**params)
else:
try:
params['value'] = json.dumps(value)
return await request.app.redis.set(**params)
except json.JSONDecodeError:
return False
async def get_cache(key: str, request: Request) -> str:
return await request.app.redis.get('blog.cache.' + key)
async def del_cache(key: str, request: Request) -> str:
return await request.app.redis.delete('blog.cache.' + key)
可是在后续的使用过程中发现每次需要调用缓存的时候必须要传入request, 也就是必须从接口进入, 非常的不优雅
于是发现redis是注册在request.app中, 于是直接引用app, 尝试使用app.redis.get/set/delete, 发现是有效的, 只是我目前的项目代码结构会导致循环引用的问题, 于是优化了项目的代码结构
3. 改为直接使用app.redis.get/set
现在就方便很多了, 在其他地方使用的时候也不需要注意传递request
import json
from datetime import timedelta
from config.init_blog import app
async def set_cache(key: str, value, ex: timedelta = None) -> bool:
if not all((key, value)):
return False
params = {'name': 'blog.cache.' + key, 'value': value}
if ex:
params['ex'] = ex
if isinstance(value, int | float | str):
return await app.redis.set(**params)
else:
try:
params['value'] = json.dumps(value)
return await app.redis.set(**params)
except json.JSONDecodeError:
return False
async def get_cache(key: str) -> str:
return await app.redis.get('blog.cache.' + key)
async def del_cache(key: str) -> str:
return await app.redis.delete('blog.cache.' + key)
4. 优化项目文件结构
之前我的init_blog 文件中的 create_app方法不仅创建了FastAPI的app, 还做了静态文件目录挂载, 允许访问的源配置, CORS中间件的配置, pg数据库注册, redis数据库注册, 模块路由的注册
在这里面做了很多创建app以外的事情, 这样非常的不优雅, 而且路由注册会引入其他模块的文件, 导致通用工具包在引入全局app的时候出现循环引用
于是把这个方法进行了拆分, create_app的时候只做app相关的配置(静态文件目录挂载, 允许访问的源配置, CORS中间件的配置), 其他的注册放到其他文件中, 这里建议把路由注册的方法放在一个独立的文件里, 因为这个方法需要导入所有的模块的router, 跟其他方法放在一起很容易出现循环引用
5. 添加几个测试api
我在common模块下的api文件里面都是用来测试的, 所以我就直接加到里面
@router.get('/redis/get/{key}')
async def test_redis_get(key: str):
return await get_cache(key)
@router.get('/redis/set/{key}')
async def test_redis_set(key: str, value: str):
await set_cache(key, value)
return await get_cache(key)
然后进入到swagger里面进行测试
这里简书一直上传图片失败, 所以就不放图片了, 大家可以自己测试
后续如果有遇到其他问题 我会继续更新
欢迎关注此代码仓库, 如果觉得有用请给我一个🌟
