import json

Условие

import json

from fastapi import APIRouter, Depends from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from redis import Redis from src.infrastructure.storage.db.models.product import ProductDB from src.infrastructure.storage.db.models.factory import FactoryDB

router = APIRouter() redis = Redis.from_url("redis://localhost:6379/0")

@router.get("/products/serials") async def get_serials_by_ids( ids: str, db: AsyncSession = Depends(get_db), ): product_ids = [int(x) for x in ids.split(",")] items = [] for product_id in product_ids: raw = redis.get(f"serial:{product_id}") if raw: items.append(json.loads(raw)) continue

    product = await db.get(ProductDB, product_id)
    factory = (
        await db.execute(select(FactoryDB).where(FactoryDB.code == product.factory))
    ).scalar_one()
    item = {
        "id": product.id,
        "serial_number": product.serial_number,
        "factory_number": factory.number,
    }
    redis.set(f"serial:{product_id}", json.dumps(item))
    items.append(item)

await db.commit()
return items

Ответ

В коде синхронный Redis блокирует async-обработчик, запросы к БД выполняются по одному ID, а отсутствие продукта приводит к ошибке. Нужны валидация входа, пакетное чтение и контролируемый кеш.

Используйте redis.asyncio.Redis и await, ограничьте количество ID и возвращайте понятную ошибку для неверных чисел. Получайте кеш через mget, недостающие товары — одним запросом с JOIN фабрики, затем восстанавливайте порядок результата, включая согласованную обработку повторных ID. Установите TTL и правило инвалидирования; ключ должен учитывать tenant и права, если данные не общие. Обрабатывайте отсутствие фабрики, повреждённый JSON и недоступность Redis. Для read-only запроса commit обычно не нужен; dependency get_db и закрытие клиентов должны быть определены отдельно.