- 在配置redis的時(shí)候遇到了一個(gè)神奇的問(wèn)題
在同一個(gè)請(qǐng)求中 使用redis可以set和get內(nèi)容
另一個(gè)請(qǐng)求中卻不能get到之前set的內(nèi)容
1. 注冊(cè)redis到app
參考了別人的項(xiàng)目和redis的官方文檔, 在app啟動(dòng)的時(shí)候增加了redis的注冊(cè)
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", )
然后在后續(xù)的項(xiàng)目中通過(guò)request.app.redis.get/set 來(lái)插入和獲得內(nèi)容
2. 增加通用工具方法
在工具類(lèi)中增加了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)
可是在后續(xù)的使用過(guò)程中發(fā)現(xiàn)每次需要調(diào)用緩存的時(shí)候必須要傳入request, 也就是必須從接口進(jìn)入, 非常的不優(yōu)雅
于是發(fā)現(xiàn)redis是注冊(cè)在request.app中, 于是直接引用app, 嘗試使用app.redis.get/set/delete, 發(fā)現(xiàn)是有效的, 只是我目前的項(xiàng)目代碼結(jié)構(gòu)會(huì)導(dǎo)致循環(huán)引用的問(wèn)題, 于是優(yōu)化了項(xiàng)目的代碼結(jié)構(gòu)
3. 改為直接使用app.redis.get/set
現(xiàn)在就方便很多了, 在其他地方使用的時(shí)候也不需要注意傳遞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. 優(yōu)化項(xiàng)目文件結(jié)構(gòu)
之前我的init_blog 文件中的 create_app方法不僅創(chuàng)建了FastAPI的app, 還做了靜態(tài)文件目錄掛載, 允許訪問(wèn)的源配置, CORS中間件的配置, pg數(shù)據(jù)庫(kù)注冊(cè), redis數(shù)據(jù)庫(kù)注冊(cè), 模塊路由的注冊(cè)
在這里面做了很多創(chuàng)建app以外的事情, 這樣非常的不優(yōu)雅, 而且路由注冊(cè)會(huì)引入其他模塊的文件, 導(dǎo)致通用工具包在引入全局app的時(shí)候出現(xiàn)循環(huán)引用
于是把這個(gè)方法進(jìn)行了拆分, create_app的時(shí)候只做app相關(guān)的配置(靜態(tài)文件目錄掛載, 允許訪問(wèn)的源配置, CORS中間件的配置), 其他的注冊(cè)放到其他文件中, 這里建議把路由注冊(cè)的方法放在一個(gè)獨(dú)立的文件里, 因?yàn)檫@個(gè)方法需要導(dǎo)入所有的模塊的router, 跟其他方法放在一起很容易出現(xiàn)循環(huán)引用
5. 添加幾個(gè)測(cè)試api
我在common模塊下的api文件里面都是用來(lái)測(cè)試的, 所以我就直接加到里面
@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)
然后進(jìn)入到swagger里面進(jìn)行測(cè)試
這里簡(jiǎn)書(shū)一直上傳圖片失敗, 所以就不放圖片了, 大家可以自己測(cè)試
后續(xù)如果有遇到其他問(wèn)題 我會(huì)繼續(xù)更新
歡迎關(guān)注此代碼倉(cāng)庫(kù), 如果覺(jué)得有用請(qǐng)給我一個(gè)??
