/
/
/
1"""Helper utilities for the cache controller."""
2
3from __future__ import annotations
4
5import asyncio
6import functools
7from collections.abc import Awaitable, Callable, Coroutine
8from copy import deepcopy
9from dataclasses import dataclass
10from typing import (
11 TYPE_CHECKING,
12 Any,
13 Concatenate,
14 ParamSpec,
15 Protocol,
16 TypeVar,
17 cast,
18 get_type_hints,
19)
20
21from music_assistant.controllers.cache.constants import (
22 BYPASS_CACHE,
23 DEFAULT_CACHE_EXPIRATION,
24 LOGGER,
25)
26from music_assistant.helpers.api import parse_value
27from music_assistant.helpers.json import SerializableType
28
29if TYPE_CHECKING:
30 from music_assistant import MusicAssistant
31
32
33class _Cacheable(Protocol):
34 """Protocol for objects that can use the @use_cache decorator."""
35
36 @property
37 def domain(self) -> str: ...
38
39 @property
40 def mass(self) -> MusicAssistant: ...
41
42
43ProviderT = TypeVar("ProviderT", bound="_Cacheable")
44P = ParamSpec("P")
45R = TypeVar("R")
46
47
48def use_cache(
49 expiration: int = DEFAULT_CACHE_EXPIRATION,
50 category: int = 0,
51 persistent: bool = False,
52 cache_checksum: str | None = None,
53 allow_bypass: bool | None = None,
54 base_class: Any = None,
55 allow_expired_cache: bool = False,
56 cache_none: bool = True,
57) -> Callable[
58 [Callable[Concatenate[ProviderT, P], Awaitable[R]]],
59 Callable[Concatenate[ProviderT, P], Coroutine[Any, Any, R]],
60]:
61 """
62 Return decorator that can be used to cache a method's result.
63
64 Concurrent callers that miss the cache on the same key share one execution and each
65 get their own copy of the result, or the one object when it cannot be copied.
66
67 :param expiration: Time in seconds the cache entry should be valid.
68 :param category: Category to group cache objects.
69 :param persistent: If True, the entry survives cache clears.
70 :param cache_checksum: Optional checksum to store with the cache object.
71 :param allow_bypass: Whether to respect the BYPASS_CACHE context variable.
72 :param base_class: If provided, reconstruct cached data using base_class.from_dict().
73 Handles both single dicts and lists of dicts automatically.
74 If not provided, falls back to type-annotation based reconstruction.
75 :param allow_expired_cache: If True, enable stale-while-revalidate. When the cached
76 entry has expired, return it immediately and trigger a background refresh that
77 re-runs the wrapped function and updates the cache. Expired entries also
78 survive the cache auto-cleanup task so they remain available as fallback data.
79 :param cache_none: Whether a None result is cached and served like any other value
80 (a negative hit, e.g. "no lyrics exist for this track"). Set to False for
81 methods where None signals a (transient) failure, so the call is retried on
82 the next invocation instead of serving a cached None.
83 """
84 if allow_bypass is None:
85 allow_bypass = not persistent
86
87 def _decorator(
88 func: Callable[Concatenate[ProviderT, P], Awaitable[R]],
89 ) -> Callable[Concatenate[ProviderT, P], Coroutine[Any, Any, R]]:
90 def _reconstruct(cachedata: Any) -> R:
91 if base_class is not None:
92 return cast("R", cachedata)
93 # fallback: reconstruct using the (memoized) return-type annotation
94 return cast("R", parse_value(func.__name__, cachedata, _resolve_return_hint(func)))
95
96 @functools.wraps(func)
97 async def wrapper(self: ProviderT, *args: P.args, **kwargs: P.kwargs) -> R:
98 cache = self.mass.cache
99 provider_id = getattr(self, "instance_id", self.domain)
100
101 # create a cache key dynamically based on the (remaining) args/kwargs
102 cache_key_parts = [func.__name__, *args]
103 for key in sorted(kwargs.keys()):
104 cache_key_parts.append(f"{key}{kwargs[key]}")
105 cache_key = ".".join(map(str, cache_key_parts))
106
107 # single lookup that returns the entry, its freshness and whether it was found;
108 # found distinguishes a stored None (a negative hit) from a cache miss. expired
109 # entries are only read (and deserialized) when this call serves stale data
110 cachedata, is_fresh, found = await cache.get_with_freshness(
111 cache_key,
112 provider=provider_id,
113 checksum=cache_checksum,
114 category=category,
115 allow_bypass=allow_bypass,
116 base_class=base_class,
117 include_expired=allow_expired_cache,
118 )
119 # a found value counts as a hit, except a None that this call opted out of
120 # caching (cache_none=False) which is treated as a miss and re-fetched
121 cache_hit = found and (cache_none or cachedata is not None)
122
123 if cache_hit and is_fresh:
124 return _reconstruct(cachedata)
125
126 def _store_task(result: R) -> Any:
127 return cache.set(
128 key=cache_key,
129 data=cast("SerializableType", result),
130 expiration=expiration,
131 provider=provider_id,
132 category=category,
133 checksum=cache_checksum,
134 persistent=persistent,
135 allow_expired_cache=allow_expired_cache,
136 )
137
138 if cache_hit and allow_expired_cache:
139 # serve stale data and refresh in the background;
140 # task_id deduplicates concurrent refreshes for the same entry
141 async def _background_refresh() -> None:
142 try:
143 result = await func(self, *args, **kwargs)
144 if cache_none or result is not None:
145 await _store_task(result)
146 except Exception:
147 LOGGER.exception(
148 "Background cache refresh failed for %s/%s",
149 provider_id,
150 cache_key,
151 )
152
153 self.mass.create_task(
154 _background_refresh(),
155 task_id=f"cache_refresh.{provider_id}.{cache_key}",
156 )
157 return _reconstruct(cachedata)
158
159 # cache miss (or expired entry without stale-while-revalidate):
160 # fetch synchronously, store in background
161 async def _fetch_and_store() -> R:
162 result = await func(self, *args, **kwargs)
163 if cache_none or result is not None:
164 self.mass.create_task(_store_task(result))
165 return result
166
167 # a caller that bypasses this method's cache asked for the backend, so it
168 # fetches alone rather than joining or publishing a flight
169 if allow_bypass and BYPASS_CACHE.get():
170 return await _fetch_and_store()
171
172 async def _flight() -> _FlightOutcome[R]:
173 # the outcome is returned rather than raised, to keep a routine failure quiet:
174 # a MediaNotFoundError out of a cached lookup is an ordinary result that
175 # callers deal with themselves, while a raising task would both draw a warning
176 # from create_task and, once every caller has gone, have asyncio report it
177 # against the shielded future
178 outcome: _FlightOutcome[R] = _FlightOutcome()
179 try:
180 outcome.result = await _fetch_and_store()
181 except Exception as err:
182 outcome.error = err
183 return outcome
184
185 # task_id folds concurrent callers for this key onto one execution
186 flight = self.mass.create_task(
187 _flight(), task_id=f"cache_flight.{provider_id}.{cache_key}"
188 )
189 if flight is asyncio.current_task():
190 # a body that calls back into itself for the same key is handed the very
191 # fetch it is running in, and awaiting that would wait on itself forever
192 return await _fetch_and_store()
193 # the shield keeps the shared task out of every caller's cancellation scope: a
194 # caller giving up cancels neither the fetch nor the callers still waiting
195 outcome = await asyncio.shield(flight)
196 if outcome.error is not None:
197 raise outcome.error
198 # the fetched objects stay behind with the flight, which keeps them pristine for
199 # the cache write; callers get a clone each because they do mutate results in
200 # place (per-user podcast resume state, for one)
201 try:
202 return deepcopy(cast("R", outcome.result))
203 except Exception as err:
204 LOGGER.warning(
205 "Cannot copy the shared result for %s/%s, callers share one object: %s",
206 provider_id,
207 cache_key,
208 err,
209 )
210 return cast("R", outcome.result)
211
212 return wrapper
213
214 return _decorator
215
216
217@dataclass(slots=True)
218class _FlightOutcome[ResultT]:
219 """Outcome of one shared fetch, handed to every caller awaiting that fetch."""
220
221 result: ResultT | None = None
222 error: Exception | None = None
223
224
225@functools.cache
226def _resolve_return_hint(func: Callable[..., Any]) -> Any:
227 """
228 Return the resolved return-type annotation of func, memoized per function.
229
230 A function's return annotation is invariant for the process lifetime, so it is resolved
231 once and cached instead of re-running get_type_hints() â which re-evaluates the PEP-563
232 string annotations â on every cache hit. Resolution stays lazy (it happens on the first
233 hit, not at decoration time), so forward-reference handling is unchanged.
234
235 :param func: The decorated function whose return-type annotation to resolve.
236 """
237 return get_type_hints(func)["return"]
238