89 lines
2.6 KiB
Python
89 lines
2.6 KiB
Python
import json
|
|
from typing import Any, Optional
|
|
|
|
from core.settings import REDIS_DEFAULT_TTL, REDIS_ENABLED, REDIS_PREFIX, REDIS_URL
|
|
|
|
try:
|
|
from redis import Redis
|
|
except Exception: # pragma: no cover
|
|
Redis = None # type: ignore
|
|
|
|
|
|
class RedisCache:
|
|
def __init__(self):
|
|
self.enabled = bool(REDIS_ENABLED and REDIS_URL and Redis is not None)
|
|
self.prefix = REDIS_PREFIX
|
|
self.default_ttl = int(REDIS_DEFAULT_TTL)
|
|
self._client: Optional["Redis"] = None
|
|
if self.enabled:
|
|
try:
|
|
self._client = Redis.from_url(REDIS_URL, decode_responses=True)
|
|
self._client.ping()
|
|
except Exception:
|
|
self.enabled = False
|
|
self._client = None
|
|
|
|
def _full_key(self, key: str) -> str:
|
|
return f"{self.prefix}:{key}"
|
|
|
|
def ping(self) -> bool:
|
|
if not self.enabled or self._client is None:
|
|
return False
|
|
try:
|
|
return bool(self._client.ping())
|
|
except Exception:
|
|
return False
|
|
|
|
def get_json(self, key: str) -> Any:
|
|
if not self.enabled or self._client is None:
|
|
return None
|
|
try:
|
|
raw = self._client.get(self._full_key(key))
|
|
if not raw:
|
|
return None
|
|
return json.loads(raw)
|
|
except Exception:
|
|
return None
|
|
|
|
def set_json(self, key: str, value: Any, ttl: Optional[int] = None) -> None:
|
|
if not self.enabled or self._client is None:
|
|
return
|
|
try:
|
|
self._client.setex(
|
|
self._full_key(key),
|
|
int(ttl if ttl is not None else self.default_ttl),
|
|
json.dumps(value, ensure_ascii=False, default=str),
|
|
)
|
|
except Exception:
|
|
return
|
|
|
|
def delete(self, *keys: str) -> None:
|
|
if not self.enabled or self._client is None:
|
|
return
|
|
full_keys = [self._full_key(key) for key in keys if str(key or "").strip()]
|
|
if not full_keys:
|
|
return
|
|
try:
|
|
self._client.delete(*full_keys)
|
|
except Exception:
|
|
return
|
|
|
|
def delete_prefix(self, prefix: str) -> None:
|
|
if not self.enabled or self._client is None:
|
|
return
|
|
pattern = self._full_key(f"{prefix}*")
|
|
try:
|
|
cursor = 0
|
|
while True:
|
|
cursor, rows = self._client.scan(cursor=cursor, match=pattern, count=200)
|
|
if rows:
|
|
self._client.delete(*rows)
|
|
if cursor == 0:
|
|
break
|
|
except Exception:
|
|
return
|
|
|
|
|
|
cache = RedisCache()
|
|
|