backend/app/core/redis.py

"""
Single source of truth for Redis connections.
Kills 24 duplicate get_redis() implementations across the codebase.

Usage:
    from app.core.redis import get_redis
    r = get_redis()
"""

import os

import redis as redis_lib

_REDIS_HOST = os.getenv("REDIS_HOST", "rmi-redis")
_REDIS_PORT = int(os.getenv("REDIS_PORT", "6379"))
_REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", "")
_REDIS_DB = int(os.getenv("REDIS_DB", "0"))

_client = None


def get_redis(decode_responses: bool = True) -> redis_lib.Redis:
    """Get or create the Redis client. Thread-safe singleton."""
    global _client
    if _client is not None:
        try:
            _client.ping()
            return _client
        except Exception:
            _client = None

    _client = redis_lib.Redis(
        host=_REDIS_HOST,
        port=_REDIS_PORT,
        password=_REDIS_PASSWORD or None,
        db=_REDIS_DB,
        decode_responses=decode_responses,
        socket_connect_timeout=3,
        socket_keepalive=True,
        health_check_interval=30,
    )
    return _client


def get_redis_async():
    """Async Redis client (for use with asyncio)."""
    import redis.asyncio as aioredis

    return aioredis.Redis(
        host=_REDIS_HOST,
        port=_REDIS_PORT,
        password=_REDIS_PASSWORD or None,
        db=_REDIS_DB,
        decode_responses=True,
        socket_connect_timeout=3,
    )


def invalidate_redis():
    """Force reconnection on next get_redis() call."""
    global _client
    _client = None