File size: 6,138 Bytes
de745bf | 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 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 | import time
import asyncio
from typing import Optional
from ..config import settings
# Redis is optional at import-time so the API runs without it.
try:
import redis.asyncio as aioredis
_HAS_REDIS = True
except Exception: # pragma: no cover - redis not installed
aioredis = None
_HAS_REDIS = False
_redis_client: Optional[object] = None
# In-memory TTL fallback cache: {key: (value, expires_at)}
_memory_cache: dict = {}
class CircuitBreaker:
"""Fail fast to the in-memory cache when Redis is slow or down.
Without this, `validate_batch` pays Redis' socket timeout (~2s) on every one
of the ~8 cache calls it makes per row — measured at 10.0s per row on a
host where the Upstash link is slow, while the engine itself needs ~70ms.
After `failure_threshold` consecutive slow/failed ops the circuit opens and
calls skip Redis entirely until `cooldown_seconds` elapse, then probe once.
"""
def __init__(self, failure_threshold: int = 3,
cooldown_seconds: float = 30.0, slow_threshold: float = 1.0):
self.failure_threshold = failure_threshold
self.cooldown_seconds = cooldown_seconds
self.slow_threshold = slow_threshold
self._failures = 0
self._open_until = 0.0
def allow(self) -> bool:
"""True when Redis may be contacted (closed, or cooldown elapsed)."""
return time.time() >= self._open_until
def record(self, success: bool, duration: float) -> None:
"""Record one op: slow ops count as failures even when they succeed."""
if success and duration < self.slow_threshold:
self._failures = 0
return
self._failures += 1
if self._failures >= self.failure_threshold:
self._open_until = time.time() + self.cooldown_seconds
self._failures = 0
def reset(self) -> None:
self._failures = 0
self._open_until = 0.0
# Trips when Redis is consistently slow/failing; see CircuitBreaker.
_circuit = CircuitBreaker()
async def get_redis():
"""Lazily create a shared Redis client (returns None if unavailable).
Redis connections must be configured explicitly. Missing or unreachable
Redis leaves the API on its bounded in-memory cache instead of silently
provisioning a temporary external database.
"""
global _redis_client
if not _HAS_REDIS:
return None
if _redis_client is None:
url = (settings.redis_url or "").strip()
if not url:
return None
try:
_redis_client = aioredis.from_url(
url,
encoding="utf-8",
decode_responses=True,
# The TLS handshake to Upstash measures ~750-850 ms on some
# networks; a 2 s cap killed the very first PING at exactly
# the timeout (observed live), which made every cache op pay
# the full budget and trip the breaker. 5 s connects reliably
# while the breaker still fails fast on real outages.
socket_connect_timeout=5,
socket_timeout=5,
)
await _redis_client.ping()
except Exception:
_redis_client = None
return _redis_client
def _memory_get(key: str) -> Optional[str]:
item = _memory_cache.get(key)
if item is None:
return None
value, expires = item
if time.time() > expires:
_memory_cache.pop(key, None)
return None
return value
def _memory_set(key: str, value: str, ttl: int):
_memory_cache[key] = (value, time.time() + ttl)
# keep cache bounded
if len(_memory_cache) > 20000:
now = time.time()
for k in [k for k, (_, exp) in _memory_cache.items() if exp < now]:
_memory_cache.pop(k, None)
async def get_cache(key: str) -> Optional[str]:
if _circuit.allow():
t0 = time.perf_counter()
try:
client = await get_redis()
if client is not None:
value = await client.get(key)
_circuit.record(success=True, duration=time.perf_counter() - t0)
return value
# Redis is configured but get_redis() could not establish a client
# (connect/ping failed). Without this the 2s timeout is paid on
# EVERY call and the breaker never trips.
if (settings.redis_url or "").strip():
_circuit.record(success=False,
duration=time.perf_counter() - t0)
except Exception:
_circuit.record(success=False, duration=time.perf_counter() - t0)
return _memory_get(key)
async def set_cache(key: str, value: str, ttl: int = 86400):
if _circuit.allow():
t0 = time.perf_counter()
try:
client = await get_redis()
if client is not None:
await client.setex(key, ttl, value)
_circuit.record(success=True, duration=time.perf_counter() - t0)
return
if (settings.redis_url or "").strip():
_circuit.record(success=False,
duration=time.perf_counter() - t0)
except Exception:
_circuit.record(success=False, duration=time.perf_counter() - t0)
_memory_set(key, value, ttl)
async def redis_available(timeout: float = 4.0) -> bool:
"""
Quick non-blocking check: can we talk to Redis?
Used to gate Celery .delay() calls so a missing Redis broker fails
fast instead of entering Celery's infinite reconnect/retry loop.
"""
if not _HAS_REDIS:
return False
try:
client = await asyncio.wait_for(
get_redis(), timeout=timeout
)
if client is None:
return False
await asyncio.wait_for(client.ping(), timeout=timeout)
return True
except asyncio.TimeoutError:
print("[REDIS] Connection timeout - Redis may not be running")
return False
except Exception as e:
print(f"[REDIS] Connection failed: {str(e)}")
return False
|