带有 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 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整个环境基于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 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())

Answer 2

您可以使用 redis_cache 来访问 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 访问：

uvicorn main:app --reload