FastAPI хранение Pydantic model в redis
Пишу приложение на FastAPI. Есть Pydantic model:
class Status(str, Enum):
inactive = 'inactive'
active = 'active'
done = 'done'
deprecated = 'deprecated'
class ShowTask(BaseModel):
tsk_id: UUID
title: constr(max_length=255)
describe: str
date_of_create: date
status: Status
class Config:
json_encoders = {
UUID: lambda v: str(v) # конвертация UUID в str при сериализации в JSON
}
from_attributes = True
Хочу хранить данные в кеше redis. Использую библиотеку aioredis. Вот коннект:
redis = aioredis.from_url(
settings.redis_url,
encoding="utf-8",
decode_responses=True,
password=settings.redis_password)
И вот endpoint, через который получаем данные из базы и потом кешим
@task_router.get("/get-task-list", response_model=list[ShowTask])
async def get_task_list(
skip: int | None = None,
limit: int | None = None,
db: AsyncSession = Depends(get_db)
):
try:
# Данные из кэша
cache = await redis.get(f"task-list.{skip}.{limit}")
if cache is not None:
try:
data = json.loads(cache)
print(data)
show_tasks = [ShowTask(**item) for item in data]
return {
"status": 200,
"from": "redis",
"data": show_tasks
}
except json.JSONDecodeError as e:
print("Ошибка при десериализации данных:", e)
# Получение данных из бд
task_list: list[ShowTask] = await service.get_multi(skip=skip, limit=limit,
session=db)
# Сохранение сериализованных данных в Redis
await redis.set(f"task-list.{skip}.{limit}", json.dumps(task_list))
return task_list
except UniqueConstraintError as err:
raise HTTPException(status_code=400, detail=str(err))
except DatabaseError as err:
raise HTTPException(status_code=500, detail=str(err))
except Exception as err:
raise HTTPException(status_code=500, detail=str(err))
Не могу получить обратно из кеша объект Pydantic model. В редис сохраняется строка с экранами. Пробовал использовать pickle, но при обратной распаковке выходит ошибка, что формат не соответствует utf-8. Как сохранять в redis питоновские объекты и получать их обратно?
Источник: Stack Overflow на русском