from json import dumps, loads
from typing import List, Optional
from pynyaata.cache import CACHE_TIMEOUT, REDIS_URL
from pynyaata.types import Cache, RemoteFile
from redis import ConnectionError, Redis
if not REDIS_URL:
raise ConnectionError(f"Invalid REDIS_URL: {REDIS_URL}")
client = Redis.from_url(REDIS_URL)
class RedisCache(Cache):
def get(self, key: str) -> Optional[List[RemoteFile]]:
data = client.get(key)
return loads(str(data)) if data else None
def set(self, key: str, data: List[RemoteFile]):
return client.set(key, dumps(data), CACHE_TIMEOUT)
