app/redis_client.py

"""Shared async Redis client for Phase 3 (rate limits, job queue)."""

from __future__ import annotations

import logging
from typing import TYPE_CHECKING

from app.config import settings

if TYPE_CHECKING:
    from redis.asyncio import Redis

logger = logging.getLogger(__name__)

_redis: Redis | None = None


def redis_configured() -> bool:
    return bool((settings.redis_url or "").strip())


def job_queue_enabled() -> bool:
    return bool(settings.enable_job_queue and redis_configured())


async def _disconnect_redis() -> None:
    global _redis
    if _redis is not None:
        try:
            await _redis.aclose()
        except Exception:  # noqa: BLE001
            pass
        _redis = None


async def get_redis() -> Redis:
    """Return a process-wide async Redis client (lazy connect, reconnect on failure)."""
    global _redis
    if _redis is not None:
        try:
            await _redis.ping()
            return _redis
        except Exception:  # noqa: BLE001
            logger.warning("Redis connection lost; reconnecting")
            await _disconnect_redis()

    if not redis_configured():
        raise RuntimeError("Redis is not configured (set REDIS_URL)")

    try:
        from redis.asyncio import Redis
    except ImportError as exc:  # pragma: no cover
        raise RuntimeError(
            "redis package is required for REDIS_URL. Install: pip install 'redis>=5.0.0'"
        ) from exc

    client = Redis.from_url(
        settings.redis_url,
        encoding="utf-8",
        decode_responses=True,
    )
    try:
        await client.ping()
    except Exception as exc:  # noqa: BLE001
        await client.aclose()
        raise RuntimeError(f"Redis ping failed for {settings.redis_url!r}: {exc}") from exc
    _redis = client
    logger.info("Redis connected url=%s", settings.redis_url.split("@")[-1])
    return _redis


async def close_redis() -> None:
    await _disconnect_redis()


async def redis_health_ok() -> bool:
    if not redis_configured():
        return False
    try:
        client = await get_redis()
        await client.ping()
        return True
    except Exception:  # noqa: BLE001
        return False


async def generation_queue_depth() -> int | None:
    """Length of the generation job queue, or None if Redis is unavailable."""
    if not redis_configured():
        return None
    try:
        client = await get_redis()
        return int(await client.llen(settings.job_queue_key))
    except Exception:  # noqa: BLE001
        return None