redis_cache.py

import hashlib
import json
import logging
import os

from dotenv import load_dotenv
from redis import Redis
from redis.exceptions import RedisError


load_dotenv()

logger = logging.getLogger(__name__)

DEFAULT_REDIS_URL = "redis://localhost:6379/0"
DEFAULT_CACHE_TTL_SECONDS = 3600


def _cache_ttl_seconds():
    value = os.getenv("CACHE_TTL_SECONDS", str(DEFAULT_CACHE_TTL_SECONDS))
    try:
        ttl = int(value)
        if ttl <= 0:
            raise ValueError
        return ttl
    except ValueError:
        logger.warning(
            "Invalid CACHE_TTL_SECONDS=%r; using %s seconds",
            value,
            DEFAULT_CACHE_TTL_SECONDS,
        )
        return DEFAULT_CACHE_TTL_SECONDS


def analysis_cache_key(keyword, since_date):
    """Build a stable key without putting arbitrary user input into Redis keys."""
    normalized_keyword = keyword.strip().lower()
    keyword_hash = hashlib.sha256(normalized_keyword.encode("utf-8")).hexdigest()
    return f"devtrends:analysis:v1:{since_date.isoformat()}:{keyword_hash}"


class RedisCache:
    def __init__(self, client=None, ttl_seconds=None, enabled=None):
        if enabled is None:
            enabled = os.getenv("CACHE_ENABLED", "true").strip().lower() not in {
                "0",
                "false",
                "no",
                "off",
            }

        self.enabled = enabled
        self.ttl_seconds = (
            ttl_seconds if ttl_seconds is not None else _cache_ttl_seconds()
        )
        self.client = client

        if self.enabled and self.client is None:
            try:
                self.client = Redis.from_url(
                    os.getenv("REDIS_URL", DEFAULT_REDIS_URL),
                    decode_responses=True,
                    socket_connect_timeout=1,
                    socket_timeout=1,
                )
            except (RedisError, ValueError) as exc:
                logger.warning(
                    "Redis cache configuration is invalid; disabling cache: %s", exc
                )
                self.enabled = False

    def get_json(self, key):
        if not self.enabled or self.client is None:
            return None

        try:
            value = self.client.get(key)
            return json.loads(value) if value is not None else None
        except (RedisError, OSError, ValueError, TypeError) as exc:
            logger.warning("Redis cache read failed; continuing without cache: %s", exc)
            return None

    def set_json(self, key, value):
        if not self.enabled or self.client is None:
            return False

        try:
            self.client.setex(key, self.ttl_seconds, json.dumps(value))
            return True
        except (RedisError, OSError, ValueError, TypeError) as exc:
            logger.warning("Redis cache write failed; continuing without cache: %s", exc)
            return False


analysis_cache = RedisCache()