src/services/cache.py

import hashlib
import json
from typing import Any, Optional

from src.core.config import REDIS_URL, REDIS_TTL_INFERENCE, REDIS_TTL_SEARCH

_redis_client = None


async def get_redis():
    global _redis_client
    if _redis_client is None and REDIS_URL:
        try:
            import redis.asyncio as aioredis
            _redis_client = aioredis.from_url(
                REDIS_URL,
                encoding="utf-8",
                decode_responses=False,
                socket_connect_timeout=2,
                socket_timeout=2,
                retry_on_timeout=True,
            )
            await _redis_client.ping()
        except Exception:
            _redis_client = None
    return _redis_client


async def cache_get(key: str) -> Optional[Any]:
    r = await get_redis()
    if not r:
        return None
    try:
        data = await r.get(key)
        return json.loads(data) if data else None
    except Exception:
        return None


async def cache_set(key: str, value: Any, ttl: int = 3600) -> bool:
    r = await get_redis()
    if not r:
        return False
    try:
        await r.setex(key, ttl, json.dumps(value, default=_numpy_serializer))
        return True
    except Exception:
        return False


async def cache_delete_pattern(pattern: str) -> int:
    r = await get_redis()
    if not r:
        return 0
    try:
        keys = await r.keys(pattern)
        if keys:
            return await r.delete(*keys)
        return 0
    except Exception:
        return 0


async def cache_flush_all() -> bool:
    r = await get_redis()
    if not r:
        return False
    try:
        await r.flushdb()
        return True
    except Exception:
        return False


def make_inference_key(image_bytes: bytes, detect_faces: bool, mode: str) -> str:
    h = hashlib.md5(image_bytes[:65536]).hexdigest()
    return f"inf:{h}:{int(detect_faces)}:{mode}"


def make_face_search_key(vec: list, threshold: float) -> str:
    h = hashlib.md5(str(vec[:8]).encode()).hexdigest()[:16]
    return f"face:{h}:{threshold:.2f}"


def make_obj_search_key(vec: list) -> str:
    h = hashlib.md5(str(vec[:8]).encode()).hexdigest()[:16]
    return f"obj:{h}"


def make_categories_key(cloud_name: str) -> str:
    return f"cats:{cloud_name}"


def _numpy_serializer(obj):
    try:
        import numpy as np
        if isinstance(obj, np.ndarray):
            return obj.tolist()
        if isinstance(obj, np.floating):
            return float(obj)
        if isinstance(obj, np.integer):
            return int(obj)
    except ImportError:
        pass
    raise TypeError(f"Object of type {type(obj)} is not JSON serializable")