带有 aiocache 和 Redis 的 FastAPI 无法设置 databases.

Question

我正在尝试使用 aiocache 库在我的端点上实施 Redis。我对 aiocache 进行的第一个测试我使用了@cache，但没有指示任何其他服务并且一切正常。但是当我尝试使用 Redis 时，我看到了这个错误（端点仍然返回请求）

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 error

Traceback (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

整个环境基于docker，2个容器，1个FastApi，1个PostgreSQL和1个Redis。

很明显端点返回的对象有问题，所以我问你如何将这么复杂的对象传递给Redis？

按照 aiochace 文档，我尝试了所有存在的序列化程序，但没有成功。

Answer 1

错误说明了一切TypeError: <databases.backends.postgres.Record object at 0x7fb9f01d8a60> is not JSON serializable错误在于您正在序列化 JSON 中的对象以缓存它，而这在 JSON 中是不可序列化的。我曾经遇到过类似的问题，但是发现fastAPI的编码器支持该Record对象，而其他的则不支持。您可以返回通过此类编码器序列化的对象，也可以在 redis 缓存参数中将其设置为编码器。我没有测试以下代码，但它应该足以说明我的意思。from fastapi.encoders import [e-mail]("/", 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())

Answer 2

您可以使用 redis_cache 来访问 RedisDBconnection.pyfrom typing import Optionalfrom aioredis import Redis, create_redis_pool#Create a RedisCache instanceclass RedisCache:&nbsp; &nbsp;&nbsp;&nbsp; &nbsp; def __init__(self):&nbsp; &nbsp; &nbsp; &nbsp; self.redis_cache: Optional[Redis] = None&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;&nbsp; &nbsp; async def init_cache(self):&nbsp; &nbsp; &nbsp; &nbsp; self.redis_cache = await create_redis_pool("redis://localhost:6379/0?encoding=utf-8") #Connecting to database&nbsp; &nbsp; async def keys(self, pattern):&nbsp; &nbsp; &nbsp; &nbsp; return await self.redis_cache.keys(pattern)&nbsp; &nbsp; async def set(self, key, value):&nbsp; &nbsp; &nbsp; &nbsp; return await self.redis_cache.set(key, value)&nbsp; &nbsp;&nbsp;&nbsp; &nbsp; async def get(self, key):&nbsp; &nbsp; &nbsp; &nbsp; return await self.redis_cache.get(key)&nbsp; &nbsp;&nbsp;&nbsp; &nbsp; async def close(self):&nbsp; &nbsp; &nbsp; &nbsp; self.redis_cache.close()&nbsp; &nbsp; &nbsp; &nbsp; await self.redis_cache.wait_closed()redis_cache = RedisCache()main.pyfrom fastapi import FastAPI, applicationsfrom uvicorn import runfrom fastapi import FastAPI, Request, Responsefrom connection import redis_cacheapp = FastAPI(title="FastAPI with Redis")async def get_all():&nbsp; &nbsp; return await redis_cache.keys('*')@app.on_event('startup')async def starup_event():&nbsp; &nbsp; await redis_cache.init_cache()@app.on_event('shutdown')async def shutdown_event():&nbsp; &nbsp; redis_cache.close()&nbsp; &nbsp; await redis_cache.wait_closed()#[e-mail]("/")def read_root():&nbsp; &nbsp; return {"Redis": "FastAPI"}#root > Get all keys from the redis [e-mail]('/RedisKeys')async def redis_keys():&nbsp; &nbsp; return await get_all()if __name__ == '__main__':&nbsp; &nbsp; run("main:app", port=3000, reload=True)我正在使用 uvicorn 访问：uvicorn main:app --reload