Codex Release Bot
97 lines
3.6 KiB
Python
97 lines
3.6 KiB
Python
"""Nexus — Redis Client (Core Data Layer)
|
|
|
|
ADR-007: Redis is the authoritative source for session/online/alert data.
|
|
ADR-008: Use BlockingConnectionPool (waits on timeout instead of throwing exception).
|
|
ADR-009: Startup mandatory validation — if Redis unavailable, exit immediately.
|
|
No `Redis | None` return type — callers never need `if redis:` guards.
|
|
"""
|
|
|
|
import logging
|
|
from fastapi import FastAPI
|
|
|
|
import redis.asyncio as aioredis
|
|
from redis.asyncio import BlockingConnectionPool
|
|
|
|
from server.config import settings
|
|
|
|
logger = logging.getLogger("nexus.redis")
|
|
|
|
_redis: aioredis.Redis | None = None
|
|
|
|
|
|
async def init_redis(app: FastAPI) -> None:
|
|
"""Initialize Redis connection during app startup (called from lifespan).
|
|
|
|
ADR-009: Redis is mandatory. If unavailable at startup, exit immediately.
|
|
No fallback mode — development environment has full stack.
|
|
"""
|
|
global _redis
|
|
|
|
if not settings.REDIS_URL:
|
|
logger.error("REDIS_URL is empty! Redis is mandatory for Nexus (ADR-007).")
|
|
raise SystemExit("REDIS_URL must be set in .env — Redis is core data layer.")
|
|
|
|
try:
|
|
pool = BlockingConnectionPool.from_url(
|
|
settings.REDIS_URL,
|
|
decode_responses=True,
|
|
max_connections=50,
|
|
timeout=5, # Wait up to 5s for available connection
|
|
health_check_interval=30, # Validate connection every 30s
|
|
retry_on_timeout=True, # Auto-retry on timeout
|
|
socket_connect_timeout=5, # 5s connection timeout
|
|
socket_timeout=5, # 5s operation timeout
|
|
)
|
|
_redis = aioredis.Redis(connection_pool=pool)
|
|
|
|
# Mandatory startup validation (ADR-009)
|
|
await _redis.ping()
|
|
# Mask credentials in URL before logging (redis://user:pwd@host → redis://**@host)
|
|
_safe_url = settings.REDIS_URL
|
|
try:
|
|
from urllib.parse import urlparse, urlunparse
|
|
_p = urlparse(settings.REDIS_URL)
|
|
if _p.password:
|
|
_p = _p._replace(netloc=f"{_p.username or ''}:**@{_p.hostname}:{_p.port or 6379}")
|
|
_safe_url = urlunparse(_p)
|
|
except Exception:
|
|
_safe_url = "redis://<masked>"
|
|
logger.info(f"Redis connected: {_safe_url} (BlockingConnectionPool, max=50)")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Redis startup validation failed: {e}")
|
|
raise SystemExit(f"Redis unavailable — Nexus cannot start (ADR-009). Error: {e}") from e
|
|
|
|
|
|
async def close_redis() -> None:
|
|
"""Close Redis connection during app shutdown"""
|
|
global _redis
|
|
if _redis:
|
|
await _redis.aclose()
|
|
_redis = None
|
|
logger.info("Redis connection closed")
|
|
|
|
|
|
def get_redis() -> aioredis.Redis:
|
|
"""Get Redis connection — always returns a valid Redis instance.
|
|
|
|
After init_redis() succeeds at startup, this always returns a connected Redis.
|
|
No `Redis | None` return type — callers never need `if redis:` guards.
|
|
|
|
Raises RuntimeError if called before init_redis() (should never happen in production).
|
|
"""
|
|
if _redis is None:
|
|
raise RuntimeError(
|
|
"Redis not initialized — call init_redis() during app startup first. "
|
|
"This should never happen in production (ADR-009)."
|
|
)
|
|
return _redis
|
|
|
|
|
|
def get_redis_sync() -> aioredis.Redis | None:
|
|
"""Get Redis connection — safe variant that returns None if not initialized.
|
|
|
|
Only used in edge cases where Redis might not be ready yet (e.g., background tasks
|
|
during graceful shutdown). Normal code should use get_redis() instead.
|
|
"""
|
|
return _redis |