import datetime from collections.abc import Callable from functools import wraps from typing import Any, Concatenate, ParamSpec, Protocol, TypeVar, cast from anydi import singleton from cachelib import BaseCache as Cache from fansifter_common.utils import timezone from fansifter_common.utils.cache import make_cache_key from dmp.config import Settings @singleton class ArtistCache: def __init__(self, cache: Cache, settings: Settings) -> None: self.cache = cache self.settings = settings def get(self, global_participant_id: str, key: str) -> Any: """Get a value from the cache.""" key = self._make_artist_key(global_participant_id, key) return self.cache.get(key) def set(self, global_participant_id: str, key: str, value: Any) -> None: """Set a value in the cache.""" key = self._make_artist_key(global_participant_id, key) timeout = self.timeout_provider() self.cache.set(key, value, timeout=timeout) def _make_artist_key(self, global_participant_id: str, key: str) -> str: return f"{self.settings.artist_cache_prefix}:{global_participant_id}:{key}" def timeout_provider(self) -> int: tomorrow = timezone.now() + datetime.timedelta(days=1) update_at = datetime.datetime( year=tomorrow.year, month=tomorrow.month, day=tomorrow.day, hour=self.settings.artist_cache_reset_at.hour, minute=self.settings.artist_cache_reset_at.minute, tzinfo=self.settings.artist_cache_reset_at.tzinfo, ) return int((update_at - timezone.now()).total_seconds()) class HasArtistCacheProtocol(Protocol): artist_cache: ArtistCache HasArtistCache = TypeVar("HasArtistCache", bound=HasArtistCacheProtocol) P = ParamSpec("P") R = TypeVar("R") def cached_artist_data( key: str, ) -> Callable[ [Callable[Concatenate[HasArtistCache, str, P], R]], Callable[Concatenate[HasArtistCache, str, P], R], ]: def decorator( func: Callable[Concatenate[HasArtistCache, str, P], R], ) -> Callable[Concatenate[HasArtistCache, str, P], R]: @wraps(func) def wrapper( self: HasArtistCache, global_participant_id: str, /, *args: P.args, **kwargs: P.kwargs, ) -> R: _key = make_cache_key(key, *args, **kwargs) value = self.artist_cache.get(global_participant_id, _key) if value is None: value = func(self, global_participant_id, *args, **kwargs) self.artist_cache.set(global_participant_id, _key, value) return cast(R, value) return wrapper return decorator