我正在嘗試使用 aiocache 庫(kù)在我的端點(diǎn)上實(shí)施 Redis。我對(duì) aiocache 進(jìn)行的第一個(gè)測(cè)試我使用了@cache，但沒(méi)有指示任何其他服務(wù)并且一切正常。但是當(dāng)我嘗試使用 Redis 時(shí)，我看到了這個(gè)錯(cuò)誤（端點(diǎn)仍然返回請(qǐng)求）ERROR: Couldn't set [<databases.backends.postgres.Record object at 0x7fb9f01d86a0>, <databases.backends.postgres.Record object at 0x7fb9f01d88b0>, <databases.backends.postgres.Record object at 0x7fb9f01d8a60>] in key app.api.authorget_authors()[], unexpected errorTraceback (most recent call last): File "/usr/local/lib/python3.8/site-packages/aiocache/decorators.py", line 144, in set_in_cache await self.cache.set(key, value, ttl=self.ttl) File "/usr/local/lib/python3.8/site-packages/aiocache/base.py", line 61, in _enabled return await func(*args, **kwargs) File "/usr/local/lib/python3.8/site-packages/aiocache/base.py", line 45, in _timeout return await asyncio.wait_for(func(self, *args, **kwargs), timeout) File "/usr/local/lib/python3.8/asyncio/tasks.py", line 483, in wait_for return fut.result() File "/us/local/lib/python3.8/site-packages/aiocache/base.py", line 75, in _plugins ret = await func(self, *args, **kwargs) File "/usr/local/lib/python3.8/site-packages/aiocache/base.py", line 265, in set ns_key, dumps(value), ttl=self._get_ttl(ttl), _cas_token=_cas_token, _conn=_conn File "/usr/local/lib/python3.8/site-packages/aiocache/serializers/serializers.py", line 140, in dumps return json.dumps(value)TypeError: <databases.backends.postgres.Record object at 0x7fb9f01d8a60> is not JSON serializable整個(gè)環(huán)境基于docker，2個(gè)容器，1個(gè)FastApi，1個(gè)PostgreSQL和1個(gè)Redis。很明顯端點(diǎn)返回的對(duì)象有問(wèn)題，所以我問(wèn)你如何將這么復(fù)雜的對(duì)象傳遞給Redis？按照 aiochace 文檔，我嘗試了所有存在的序列化程序，但沒(méi)有成功。
2 回答
回首憶惘然
TA貢獻(xiàn)1847條經(jīng)驗(yàn) 獲得超11個(gè)贊
錯(cuò)誤說(shuō)明了一切
TypeError:?<databases.backends.postgres.Record?object?at?0x7fb9f01d8a60>?is?not?JSON?serializable
錯(cuò)誤在于您正在序列化 JSON 中的對(duì)象以緩存它，而這在 JSON 中是不可序列化的。
我曾經(jīng)遇到過(guò)類(lèi)似的問(wèn)題，但是發(fā)現(xiàn)fastAPI的編碼器支持該Record對(duì)象，而其他的則不支持。您可以返回通過(guò)此類(lèi)編碼器序列化的對(duì)象，也可以在 redis 緩存參數(shù)中將其設(shè)置為編碼器。
我沒(méi)有測(cè)試以下代碼，但它應(yīng)該足以說(shuō)明我的意思。
from fastapi.encoders import jsonable_encoder
@authors.get("/", response_model=List[AuthorOut])? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ??
@cached(? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ??
? ? ?ttl=100,? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ??
? ? ?cache=Cache.REDIS,
? ? ?endpoint="X.X.X.X", #my local ip
? ? ?serializer=JsonSerializer(),
? ? ?port=6379,
? ? ?namespace="main",
? ? ?#key="key",
?)
async def get_authors():
? ?return jsonable_encoder(await db_manager.get_all_authors())
斯蒂芬大帝
TA貢獻(xiàn)1827條經(jīng)驗(yàn) 獲得超8個(gè)贊
您可以使用 redis_cache 來(lái)訪(fǎng)問(wèn) RedisDB
connection.py
from typing import Optional
from aioredis import Redis, create_redis_pool
#Create a RedisCache instance
class RedisCache:
def __init__(self):
self.redis_cache: Optional[Redis] = None
async def init_cache(self):
self.redis_cache = await create_redis_pool("redis://localhost:6379/0?encoding=utf-8") #Connecting to database
async def keys(self, pattern):
return await self.redis_cache.keys(pattern)
async def set(self, key, value):
return await self.redis_cache.set(key, value)
async def get(self, key):
return await self.redis_cache.get(key)
async def close(self):
self.redis_cache.close()
await self.redis_cache.wait_closed()
redis_cache = RedisCache()
main.py
from fastapi import FastAPI, applications
from uvicorn import run
from fastapi import FastAPI, Request, Response
from connection import redis_cache
app = FastAPI(title="FastAPI with Redis")
async def get_all():
return await redis_cache.keys('*')
@app.on_event('startup')
async def starup_event():
await redis_cache.init_cache()
@app.on_event('shutdown')
async def shutdown_event():
redis_cache.close()
await redis_cache.wait_closed()
#root
@app.get("/")
def read_root():
return {"Redis": "FastAPI"}
#root > Get all keys from the redis DB
@app.get('/RedisKeys')
async def redis_keys():
return await get_all()
if __name__ == '__main__':
run("main:app", port=3000, reload=True)
我正在使用 uvicorn 訪(fǎng)問(wèn)：
uvicorn main:app --reload
添加回答
舉報(bào)
0/150
提交
取消