| """In-memory cache for Short Hunter datasource responses.""" |
|
|
| from __future__ import annotations |
|
|
| import os |
| import threading |
| import time |
| from copy import deepcopy |
| from typing import Any, Dict, Optional |
|
|
|
|
| def _ttl_seconds() -> int: |
| return int(os.getenv("PROVIDER_CACHE_TTL_SECONDS", "30")) |
|
|
|
|
| def _stale_seconds() -> int: |
| return int(os.getenv("PROVIDER_STALE_CACHE_THRESHOLD_SECONDS", "300")) |
|
|
|
|
| class GatewayCache: |
| def __init__(self) -> None: |
| self._lock = threading.Lock() |
| self._items: Dict[str, Dict[str, Any]] = {} |
|
|
| def set(self, key: str, payload: Dict[str, Any], *, provider: str, capability: str, ttl: Optional[int] = None) -> None: |
| now = time.time() |
| ttl_seconds = int(ttl or _ttl_seconds()) |
| with self._lock: |
| self._items[key] = { |
| "key": key, |
| "provider": provider, |
| "capability": capability, |
| "storedAt": now, |
| "expiresAt": now + ttl_seconds, |
| "staleAfter": now + _stale_seconds(), |
| "payload": deepcopy(payload), |
| } |
|
|
| def get(self, key: str, *, allow_stale: bool = True) -> Optional[Dict[str, Any]]: |
| now = time.time() |
| with self._lock: |
| item = self._items.get(key) |
| if not item: |
| return None |
| if not allow_stale and now > item["expiresAt"]: |
| return None |
| result = deepcopy(item) |
| result["ageSeconds"] = max(0.0, now - item["storedAt"]) |
| result["isStale"] = now > item["staleAfter"] |
| result["isExpired"] = now > item["expiresAt"] |
| return result |
|
|
| def snapshot(self) -> Dict[str, Any]: |
| now = time.time() |
| with self._lock: |
| items = [] |
| for item in self._items.values(): |
| cloned = dict(item) |
| cloned["ageSeconds"] = max(0.0, now - item["storedAt"]) |
| cloned["isStale"] = now > item["staleAfter"] |
| cloned["isExpired"] = now > item["expiresAt"] |
| items.append(cloned) |
| return { |
| "size": len(items), |
| "ttlSeconds": _ttl_seconds(), |
| "staleThresholdSeconds": _stale_seconds(), |
| "items": items, |
| } |
|
|
|
|
| gateway_cache = GatewayCache() |
|
|
|
|
