app/core/redis_client.py

"""Shared Redis connection for distributed rate limiting and prompt cache.

Env-gated: when ``REDIS_URL`` is unset or Redis is unreachable, ``get_redis()``
returns ``None`` and every caller falls back to its in-process implementation.
This keeps the app fully functional on a single instance and upgrades to
horizontally-correct shared state the moment a ``REDIS_URL`` is provided
(e.g. an Upstash free-tier database).
"""

from __future__ import annotations

import logging
import threading
from typing import Any

from app.core.config import get_settings

logger = logging.getLogger(__name__)

_client: Any | None = None
_resolved = False
_lock = threading.Lock()


def get_redis() -> Any | None:
    """Return a connected Redis client, or ``None`` if unavailable.

    Resolved once per process. A failed connection is cached as ``None`` so a
    missing/broken Redis never adds per-request latency.
    """
    global _client, _resolved
    if _resolved:
        return _client

    with _lock:
        if _resolved:
            return _client
        _resolved = True
        settings = get_settings()
        url = (getattr(settings, "redis_url", "") or "").strip()
        if not url:
            _client = None
            return None
        try:
            import redis  # redis-py, sync client

            client = redis.Redis.from_url(
                url,
                socket_connect_timeout=2,
                socket_timeout=2,
                decode_responses=True,
                health_check_interval=30,
            )
            client.ping()
            _client = client
            logger.info("Redis connected — distributed rate-limit + cache active.")
        except Exception as exc:  # ImportError, connection, auth, timeout
            _client = None
            logger.warning("Redis unavailable (%s) — using in-memory fallback.", type(exc).__name__)
        return _client


def redis_enabled() -> bool:
    return get_redis() is not None