62 lines
1.6 KiB
Python
62 lines
1.6 KiB
Python
"""Redis client wrapper for caching and task queuing."""
|
|
|
|
import json
|
|
import logging
|
|
from typing import Optional, Any
|
|
|
|
import redis
|
|
|
|
from config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_redis_client: Optional[redis.Redis] = None
|
|
|
|
|
|
def get_redis_client() -> redis.Redis:
|
|
"""Return a singleton Redis client instance."""
|
|
global _redis_client
|
|
if _redis_client is None:
|
|
_redis_client = redis.from_url(settings.redis_url, decode_responses=True)
|
|
return _redis_client
|
|
|
|
|
|
def ping() -> bool:
|
|
"""Check Redis connectivity."""
|
|
try:
|
|
return get_redis_client().ping()
|
|
except redis.ConnectionError as exc:
|
|
logger.error("Redis ping failed: %s", exc)
|
|
return False
|
|
|
|
|
|
def set_json(key: str, value: Any, expire: Optional[int] = None) -> None:
|
|
"""Store a JSON-serializable value in Redis."""
|
|
client = get_redis_client()
|
|
try:
|
|
client.set(key, json.dumps(value), ex=expire)
|
|
except redis.RedisError as exc:
|
|
logger.error("Redis set_json failed: %s", exc)
|
|
raise
|
|
|
|
|
|
def get_json(key: str) -> Optional[Any]:
|
|
"""Retrieve and deserialize a JSON value from Redis."""
|
|
client = get_redis_client()
|
|
try:
|
|
data = client.get(key)
|
|
return json.loads(data) if data is not None else None
|
|
except redis.RedisError as exc:
|
|
logger.error("Redis get_json failed: %s", exc)
|
|
raise
|
|
|
|
|
|
def delete_key(key: str) -> int:
|
|
"""Delete a key from Redis. Returns number of deleted keys."""
|
|
client = get_redis_client()
|
|
try:
|
|
return client.delete(key)
|
|
except redis.RedisError as exc:
|
|
logger.error("Redis delete_key failed: %s", exc)
|
|
raise
|