Spaces:
Runtime error
Runtime error
File size: 2,530 Bytes
732b14f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | """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
|