cache.py

"""A tiny caching helper backed by Redis, with a safe in-memory fallback.

The fallback means the app and tests run even when Redis isn't available. In
production you'd rely on Redis so the cache is shared across worker processes.
"""
import json
from typing import Any, Optional

try:
    import redis  # type: ignore

    from app.config import settings

    _client: Optional["redis.Redis"] = redis.Redis.from_url(
        settings.redis_url, decode_responses=True
    )
except Exception:  # redis not installed or unreachable
    _client = None

# In-memory fallback store.
_memory: dict[str, str] = {}


def cache_get(key: str) -> Optional[Any]:
    raw: Optional[str]
    if _client is not None:
        try:
            raw = _client.get(key)
        except Exception:
            raw = _memory.get(key)
    else:
        raw = _memory.get(key)
    return json.loads(raw) if raw else None


def cache_set(key: str, value: Any, ttl_seconds: int = 60) -> None:
    raw = json.dumps(value, default=str)
    if _client is not None:
        try:
            _client.set(key, raw, ex=ttl_seconds)
            return
        except Exception:
            pass
    _memory[key] = raw


def cache_delete(key: str) -> None:
    if _client is not None:
        try:
            _client.delete(key)
        except Exception:
            pass
    _memory.pop(key, None)