import copy
import time
from typing import Union
from collections import OrderedDict
import requests
from yandex_music.utils.request import Request as YandexMusicRequest
class LimitedSizeDict(OrderedDict):
def __init__(self, *args, **kwargs):
self.size_limit = kwargs.pop("size_limit", None)
OrderedDict.__init__(self, *args, **kwargs)
self._check_size_limit()
def __setitem__(self, key, value):
OrderedDict.__setitem__(self, key, value)
self._check_size_limit()
def _check_size_limit(self):
if self.size_limit is not None:
while len(self) > self.size_limit:
self.popitem(last=False)
class CachedItem:
def __init__(self, response: requests.Response):
self.timestamp = time.time()
self.response = response
class Cache:
__singleton: 'Cache' = None
def __init__(self, lifetime: Union[str, int], size: Union[str, int]):
Cache.__singleton = self
self.lifetime = int(lifetime) * 60
self.size = int(size)
self.storage: LimitedSizeDict = LimitedSizeDict(size_limit=int(size))
def get(self, *args, **kwargs):
hash_ = Cache.get_hash(*args, **kwargs)
return self.storage[hash_].response
def update(self, response: requests.Response, *args, **kwargs):
hash_ = Cache.get_hash(*args, **kwargs)
self.storage.update({hash_: CachedItem(response)})
def need_to_fetch(self, *args, **kwargs):
hash_ = Cache.get_hash(*args, **kwargs)
cached_item = self.storage.get(hash_)
if not cached_item:
return True
if time.time() - cached_item.timestamp > self.lifetime:
return True
return False
@classmethod
def get_instance(cls) -> 'Cache':
if cls.__singleton is not None:
return cls.__singleton
else:
raise RuntimeError(f'{cls.__name__} not initialized')
@staticmethod
def freeze_dict(d: dict) -> Union[frozenset, tuple, dict]:
if isinstance(d, dict):
return frozenset((key, Cache.freeze_dict(value)) for key, value in d.items())
elif isinstance(d, list):
return tuple(Cache.freeze_dict(value) for value in d)
return d
@staticmethod
def get_hash(*args, **kwargs):
return hash((args, Cache.freeze_dict(kwargs)))
class Request(YandexMusicRequest):
def _request_wrapper(self, *args, **kwargs):
use_cache = kwargs.get('use_cache', True)
if 'use_cache' in kwargs:
kwargs.pop('use_cache')
if not use_cache:
response = super()._request_wrapper(*args, **copy.deepcopy(kwargs))
elif use_cache and Cache.get_instance().need_to_fetch(*args, **kwargs):
response = super()._request_wrapper(*args, **copy.deepcopy(kwargs))
Cache.get_instance().update(response, *args, **kwargs)
else:
response = Cache.get_instance().get(*args, **kwargs)
return response