redis_cache.py

"""Lightweight Redis cache helper. Falls through gracefully if Redis is unavailable."""
import hashlib
import json
import logging
import os
from typing import Any, Optional

try:
    import redis as _redis_pkg
except ImportError:
    _redis_pkg = None

log = logging.getLogger(__name__)

REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379")
PREFIX = os.environ.get("REDIS_KEY_PREFIX", "reco:")

_client = None
_disabled = False


def _get_client():
    global _client, _disabled
    if _disabled or _redis_pkg is None:
        return None
    if _client is None:
        try:
            _client = _redis_pkg.from_url(
                REDIS_URL,
                socket_connect_timeout=2,
                socket_timeout=2,
                decode_responses=False,
            )
            _client.ping()
            log.info("Redis cache connected: %s", REDIS_URL)
        except Exception as e:
            log.warning("Redis cache disabled: %s", e)
            _disabled = True
            _client = None
    return _client


def _k(key: str) -> str:
    return key if key.startswith(PREFIX) else PREFIX + key


def cache_get(key: str) -> Optional[Any]:
    c = _get_client()
    if c is None:
        return None
    try:
        raw = c.get(_k(key))
        return json.loads(raw) if raw else None
    except Exception as e:
        log.warning("cache_get failed (%s): %s", key, e)
        return None


def cache_set(key: str, value: Any, ttl_sec: int) -> None:
    c = _get_client()
    if c is None:
        return
    try:
        c.set(_k(key), json.dumps(value), ex=ttl_sec)
    except Exception as e:
        log.warning("cache_set failed (%s): %s", key, e)


def cache_del(pattern: str) -> int:
    c = _get_client()
    if c is None:
        return 0
    full = _k(pattern)
    try:
        if "*" not in full:
            return c.delete(full)
        deleted = 0
        cursor = 0
        while True:
            cursor, keys = c.scan(cursor=cursor, match=full, count=200)
            if keys:
                deleted += c.delete(*keys)
            if cursor == 0:
                break
        return deleted
    except Exception as e:
        log.warning("cache_del failed (%s): %s", pattern, e)
        return 0


def hash_str(s: str) -> str:
    return hashlib.sha256(s.encode("utf-8")).hexdigest()