/
/
/
1"""Apple Music API client."""
2
3from __future__ import annotations
4
5import functools
6from collections.abc import AsyncGenerator, Awaitable, Callable
7from typing import TYPE_CHECKING, Any, Concatenate, cast
8
9from aiohttp import ClientConnectionError, ClientPayloadError, ClientTimeout
10from music_assistant_models.enums import MediaType
11from music_assistant_models.errors import (
12 LoginFailed,
13 MediaNotFoundError,
14 RateLimited,
15 ResourceTemporarilyUnavailable,
16)
17
18from music_assistant.helpers.json import json_loads
19from music_assistant.helpers.throttle_retry import ThrottlerManager, throttle_with_retries
20
21from .helpers.utils import is_library_id, translate_media_type_to_apple_type
22
23if TYPE_CHECKING:
24 from .provider import AppleMusicProvider
25
26_APPLE_API_BASE = "https://api.music.apple.com/v1"
27
28_LIBRARY_PAGE_SIZE = 100
29
30_PAGE_TRUNCATION_RETRIES = 3
31
32
33def _retry_transient_transport_errors[ClientT, **P, R](
34 func: Callable[Concatenate[ClientT, P], Awaitable[R]],
35) -> Callable[Concatenate[ClientT, P], Awaitable[R]]:
36 """
37 Convert transient aiohttp transport errors into the retryable error type.
38
39 A dropped connection or truncated body raises an ``aiohttp.ClientError`` (or
40 ``TimeoutError``) rather than an HTTP status, so it would otherwise bypass the
41 status-based retry handling. ``ClientResponseError`` (raised by
42 ``raise_for_status`` for genuine 4xx/5xx) is deliberately not caught here.
43 """
44
45 @functools.wraps(func)
46 async def wrapper(self: ClientT, *args: P.args, **kwargs: P.kwargs) -> R:
47 try:
48 return await func(self, *args, **kwargs)
49 except (ClientConnectionError, ClientPayloadError, TimeoutError) as err:
50 raise ResourceTemporarilyUnavailable(
51 f"Transient transport error calling Apple Music: {type(err).__name__}: {err}"
52 ) from err
53
54 return wrapper
55
56
57def _raise_on_auth_error(status: int, endpoint: str) -> None:
58 """
59 Raise LoginFailed when Apple rejected our credentials.
60
61 Apple answers a revoked or expired music user token with 401/403 on every endpoint,
62 so this surfaces as an auth error the user can act on instead of a bare HTTP error.
63
64 :param status: The HTTP status code returned by Apple.
65 :param endpoint: The endpoint that was called, used in the error message.
66 """
67 if status in (401, 403):
68 raise LoginFailed(
69 f"Apple Music denied access to {endpoint}: the account needs to be signed in again"
70 )
71
72
73class AppleMusicAPIClient:
74 """Handles all HTTP communication with the Apple Music API."""
75
76 # period=0.25 -> 4 req/s. Apple throttles per developer account, so 429s follow the
77 # fleet-wide load on the bundled token, not our rate - and they clear within a second.
78 # 8 attempts keep the 1/2/4/8/16/32/64s ladder spanning ~2 minutes, so a sustained
79 # throttle (or outage) is still ridden out instead of failing the request in ~15s.
80 throttler = ThrottlerManager(rate_limit=1, period=0.25, retry_attempts=8, initial_backoff=1)
81
82 def __init__(self, provider: AppleMusicProvider) -> None:
83 """Initialize the API client."""
84 self.provider = provider
85 self.logger = provider.logger
86
87 @property
88 def _headers(self) -> dict[str, str]:
89 """Return standard auth headers."""
90 return {
91 "Authorization": f"Bearer {self.provider._music_app_token}",
92 "Music-User-Token": cast("str", self.provider._music_user_token),
93 }
94
95 @throttle_with_retries
96 @_retry_transient_transport_errors
97 async def get_data(self, endpoint: str, **kwargs: Any) -> dict[str, Any]:
98 """GET data from the Apple Music API."""
99 url = f"{_APPLE_API_BASE}/{endpoint}"
100 async with (
101 self.provider.mass.http_session.get(
102 url,
103 headers=self._headers,
104 params=kwargs,
105 ssl=True,
106 timeout=ClientTimeout(total=120),
107 ) as response,
108 ):
109 _raise_on_auth_error(response.status, endpoint)
110 if response.status == 404 and "limit" in kwargs and "offset" in kwargs:
111 return {}
112 if response.status == 404:
113 raise MediaNotFoundError(f"{endpoint} not found")
114 if response.status == 504:
115 self.provider.logger.debug(
116 "Apple Music API Timeout: url=%s, params=%s, response_headers=%s",
117 url,
118 kwargs,
119 response.headers,
120 )
121 raise ResourceTemporarilyUnavailable("Apple Music API Timeout")
122 if response.status == 429:
123 self.provider.logger.debug(
124 "Apple Music Rate Limiter. Headers: %s", response.headers
125 )
126 raise RateLimited("Apple Music Rate Limiter")
127 if response.status == 500:
128 # Apple 500s are typically transient; retry rather than abort the whole sync.
129 raise ResourceTemporarilyUnavailable(
130 "Unexpected server error when calling Apple Music"
131 )
132 response.raise_for_status()
133 return cast("dict[str, Any]", await response.json(loads=json_loads))
134
135 @throttle_with_retries
136 async def delete_data(self, endpoint: str, data: Any = None, **kwargs: Any) -> None:
137 """DELETE data from the Apple Music API."""
138 url = f"{_APPLE_API_BASE}/{endpoint}"
139 async with (
140 self.provider.mass.http_session.delete(
141 url,
142 headers=self._headers,
143 params=kwargs,
144 json=data,
145 ssl=True,
146 timeout=ClientTimeout(total=120),
147 ) as response,
148 ):
149 _raise_on_auth_error(response.status, endpoint)
150 if response.status == 404:
151 raise MediaNotFoundError(f"{endpoint} not found")
152 if response.status == 429:
153 self.provider.logger.debug(
154 "Apple Music Rate Limiter. Headers: %s", response.headers
155 )
156 raise RateLimited("Apple Music Rate Limiter")
157 response.raise_for_status()
158
159 @throttle_with_retries
160 async def put_data(self, endpoint: str, data: Any = None, **kwargs: Any) -> dict[str, Any]:
161 """PUT data to the Apple Music API."""
162 url = f"{_APPLE_API_BASE}/{endpoint}"
163 async with (
164 self.provider.mass.http_session.put(
165 url,
166 headers=self._headers,
167 params=kwargs,
168 json=data,
169 ssl=True,
170 timeout=ClientTimeout(total=120),
171 ) as response,
172 ):
173 _raise_on_auth_error(response.status, endpoint)
174 if response.status == 404:
175 raise MediaNotFoundError(f"{endpoint} not found")
176 if response.status == 429:
177 self.provider.logger.debug(
178 "Apple Music Rate Limiter. Headers: %s", response.headers
179 )
180 raise RateLimited("Apple Music Rate Limiter")
181 response.raise_for_status()
182 if response.content_length:
183 return cast("dict[str, Any]", await response.json(loads=json_loads))
184 return {}
185
186 @throttle_with_retries
187 async def post_data(self, endpoint: str, data: Any = None, **kwargs: Any) -> dict[str, Any]:
188 """POST data to the Apple Music API."""
189 url = f"{_APPLE_API_BASE}/{endpoint}"
190 async with (
191 self.provider.mass.http_session.post(
192 url,
193 headers=self._headers,
194 params=kwargs,
195 json=data,
196 ssl=True,
197 timeout=ClientTimeout(total=120),
198 ) as response,
199 ):
200 _raise_on_auth_error(response.status, endpoint)
201 if response.status == 404:
202 raise MediaNotFoundError(f"{endpoint} not found")
203 if response.status == 429:
204 self.provider.logger.debug(
205 "Apple Music Rate Limiter. Headers: %s", response.headers
206 )
207 raise RateLimited("Apple Music Rate Limiter")
208 response.raise_for_status()
209 return cast("dict[str, Any]", await response.json(loads=json_loads))
210
211 async def iter_all_items(
212 self, endpoint: str, key: str = "data", page_size: int = _LIBRARY_PAGE_SIZE, **kwargs: Any
213 ) -> AsyncGenerator[dict[str, Any]]:
214 """
215 Yield items from a paged list one page at a time.
216
217 Unlike :meth:`get_all_items`, this never holds the full result set in memory, so it is
218 safe for very large listings (e.g. a 100k-track Apple Music library).
219 """
220 offset = 0
221 while True:
222 kwargs["limit"] = page_size
223 kwargs["offset"] = offset
224 result = await self._get_page(endpoint, key, offset, kwargs)
225 if key not in result:
226 # offset 0 only: empty collection or 404. _get_page raises on mid-list truncation.
227 break
228 for item in result[key]:
229 yield item
230 if not result.get("next"):
231 break
232 offset += page_size
233
234 async def get_all_items(
235 self, endpoint: str, key: str = "data", **kwargs: Any
236 ) -> list[dict[str, Any]]:
237 """Get all items from a paged list."""
238 return [item async for item in self.iter_all_items(endpoint, key, **kwargs)]
239
240 async def get_user_storefront(self) -> str:
241 """Return the user's storefront identifier."""
242 locale = self.provider.mass.metadata.locale.replace("_", "-")
243 language = locale.split("-")[0]
244 result = await self.get_data("me/storefront", l=language)
245 return cast("str", result["data"][0]["id"])
246
247 async def get_ratings(self, item_ids: list[str], media_type: MediaType) -> dict[str, bool]:
248 """Return a mapping of item_id â is_favourite for a list of IDs."""
249 if media_type == MediaType.ARTIST:
250 raise NotImplementedError(
251 "Ratings are not available for artist in the Apple Music API."
252 )
253 if not item_ids:
254 return {}
255 apple_type = translate_media_type_to_apple_type(media_type)
256 endpoint = apple_type if not is_library_id(item_ids[0]) else f"library-{apple_type}"
257 max_ids_per_request = 200
258 results: dict[str, bool] = {}
259 for i in range(0, len(item_ids), max_ids_per_request):
260 batch_ids = item_ids[i : i + max_ids_per_request]
261 response = await self.get_data(
262 f"me/ratings/{endpoint}",
263 ids=",".join(batch_ids),
264 )
265 results.update(
266 {
267 item["id"]: bool(item["attributes"].get("value", False) == 1)
268 for item in response.get("data", [])
269 }
270 )
271 return results
272
273 async def _get_page(
274 self, endpoint: str, key: str, offset: int, kwargs: dict[str, Any]
275 ) -> dict[str, Any]:
276 """
277 Fetch a single page of a paged listing, recovering from transient truncation.
278
279 Apple returns ``{}`` (HTTP 404) for a paged request that yields no payload. On the
280 first page (offset 0) that legitimately means an empty collection. Mid-pagination
281 (offset > 0, reached only after a prior page promised more via ``next``) it means a
282 transient truncation; accepting it would drop still-present items and trigger
283 spurious deletions, so re-fetch the same page a bounded number of times before
284 surfacing the failure.
285 """
286 result = await self.get_data(endpoint, **kwargs)
287 if key in result or offset == 0:
288 return result
289 for _ in range(_PAGE_TRUNCATION_RETRIES):
290 result = await self.get_data(endpoint, **kwargs)
291 if key in result:
292 return result
293 raise ResourceTemporarilyUnavailable(
294 f"Incomplete paged listing for {endpoint} at offset {offset}"
295 )
296