/
/
/
1"""API Client for Tidal."""
2
3from __future__ import annotations
4
5import json
6from typing import TYPE_CHECKING, Any
7from uuid import uuid4
8
9from music_assistant_models.errors import (
10 LoginFailed,
11 MediaNotFoundError,
12 RateLimited,
13 ResourceTemporarilyUnavailable,
14)
15
16from music_assistant.helpers.throttle_retry import ThrottlerManager, throttle_with_retries
17
18from .constants import BASE_URL, JSONAPI_CONTENT_TYPE, OPEN_API_URL
19from .jsonapi import JsonApiDocument
20
21if TYPE_CHECKING:
22 from collections.abc import AsyncGenerator, Sequence
23
24 from aiohttp import ClientResponse
25
26 from .provider import TidalProvider
27
28# Safety ceiling for cursor pagination. The JSON:API relationship endpoints only
29# expose page[cursor] (no page size control), so the server fixes a small page
30# size and this cap is the only guard against a runaway cursor. It must stay high
31# enough to walk an entire library/playlist without truncating; only a genuinely
32# broken cursor should ever reach it.
33MAX_PAGINATION_PAGES = 1000
34
35
36class TidalAPIClient:
37 """Client for interacting with Tidal API."""
38
39 # Define throttler here for use by the client
40 # Rate empirically verified (2026-07): a 10-minute soak at 4/s (2400 mixed
41 # requests) plus bursts to 12/s completed without a single 429.
42 throttler = ThrottlerManager(rate_limit=4, period=1)
43
44 def __init__(self, provider: TidalProvider):
45 """Initialize API client."""
46 self.provider = provider
47 self.auth = provider.auth
48 self.logger = provider.logger
49 self.mass = provider.mass
50
51 async def get(self, endpoint: str, **kwargs: Any) -> dict[str, Any]:
52 """Get data from Tidal API."""
53 data, _ = await self._request("GET", endpoint, **kwargs)
54 return data
55
56 async def get_jsonapi(
57 self,
58 endpoint: str,
59 include: Sequence[str] | None = None,
60 params: dict[str, Any] | None = None,
61 replace_media: str | None = None,
62 **kwargs: Any,
63 ) -> JsonApiDocument:
64 """
65 Get a JSON:API document from the official Tidal API.
66
67 :param endpoint: Path below the API root.
68 :param include: Relationship paths to side-load into `included`.
69 :param params: Extra query parameters.
70 :param replace_media: Relationship path(s) whose media identifiers Tidal should
71 project onto their live replacements. Tidal churns tracks (deletes and
72 re-adds them under new ids), and this makes it hand back the live id plus
73 the original in `meta.replacement`, instead of an id that 404s.
74 """
75 query = dict(params or {})
76 if include:
77 query["include"] = ",".join(include)
78 if replace_media:
79 query["replaceMedia"] = replace_media
80 headers = kwargs.pop("headers", {})
81 headers["Accept"] = JSONAPI_CONTENT_TYPE
82 data = await self.get(
83 endpoint, base_url=OPEN_API_URL, params=query, headers=headers, **kwargs
84 )
85 # A valid JSON:API read always carries a top-level "data"; an empty body
86 # ({"success": True}) or error document lacks it, so raise instead of
87 # returning a silently empty result that would be cached as a no-match.
88 if "data" not in data:
89 raise ResourceTemporarilyUnavailable(f"Invalid JSON:API response for {endpoint}")
90 return JsonApiDocument(data)
91
92 async def write_jsonapi(
93 self, method: str, endpoint: str, body: dict[str, Any]
94 ) -> dict[str, Any]:
95 """Send a JSON:API write (POST/DELETE) to the official Tidal API."""
96 headers = {
97 "Content-Type": JSONAPI_CONTENT_TYPE,
98 "Accept": JSONAPI_CONTENT_TYPE,
99 "Idempotency-Key": str(uuid4()),
100 }
101 result, _ = await self._request(
102 method, endpoint, base_url=OPEN_API_URL, data=json.dumps(body), headers=headers
103 )
104 return result
105
106 async def get_with_etag(self, endpoint: str, **kwargs: Any) -> tuple[dict[str, Any], str]:
107 """Get data from the (unofficial) Tidal API, returning the response ETag as well."""
108 return await self._request("GET", endpoint, **kwargs)
109
110 async def post(
111 self,
112 endpoint: str,
113 data: dict[str, Any] | None = None,
114 as_form: bool = False,
115 **kwargs: Any,
116 ) -> dict[str, Any]:
117 """Send POST data to the (unofficial) Tidal API."""
118 if as_form:
119 kwargs.setdefault("headers", {})["Content-Type"] = "application/x-www-form-urlencoded"
120 kwargs["data"] = data
121 else:
122 kwargs["json"] = data
123 result, _ = await self._request("POST", endpoint, **kwargs)
124 return result
125
126 async def delete(
127 self, endpoint: str, data: dict[str, Any] | None = None, **kwargs: Any
128 ) -> dict[str, Any]:
129 """Delete data from the (unofficial) Tidal API."""
130 kwargs["json"] = data
131 result, _ = await self._request("DELETE", endpoint, **kwargs)
132 return result
133
134 async def paginate_jsonapi(
135 self,
136 endpoint: str,
137 include: Sequence[str] | None = None,
138 params: dict[str, Any] | None = None,
139 max_pages: int = MAX_PAGINATION_PAGES,
140 replace_media: str | None = None,
141 **kwargs: Any,
142 ) -> AsyncGenerator[JsonApiDocument]:
143 """Yield successive JSON:API document pages, following the cursor links."""
144 cursor: str | None = None
145 seen_cursors: set[str] = set()
146 for _ in range(max_pages):
147 page_params = dict(params or {})
148 if cursor:
149 page_params["page[cursor]"] = cursor
150 doc = await self.get_jsonapi(
151 endpoint,
152 include=include,
153 params=page_params,
154 replace_media=replace_media,
155 **kwargs,
156 )
157 yield doc
158 cursor = doc.next_cursor
159 if not cursor:
160 return
161 # A server re-serving an already-followed cursor would loop all the way
162 # to the page cap before warning; stop at the first repeat instead.
163 if cursor in seen_cursors:
164 self.logger.warning(
165 "Stopped paginating %s: server repeated cursor %s", endpoint, cursor
166 )
167 return
168 seen_cursors.add(cursor)
169 # Reached the page cap while more pages remained: surface the truncation.
170 self.logger.warning(
171 "Stopped paginating %s after %d pages; results may be truncated", endpoint, max_pages
172 )
173
174 @throttle_with_retries
175 async def _request(
176 self, method: str, endpoint: str, **kwargs: Any
177 ) -> tuple[dict[str, Any], str]:
178 """Handle API requests internally."""
179 if not await self.auth.ensure_valid_token():
180 raise LoginFailed("Failed to authenticate with Tidal")
181
182 # Prepare URL
183 base_url = kwargs.pop("base_url", BASE_URL)
184 url = f"{base_url}/{endpoint}"
185
186 # Prepare Headers
187 headers = kwargs.pop("headers", {})
188 headers["Authorization"] = f"Bearer {self.auth.access_token}"
189
190 locale = self.mass.metadata.locale.replace("_", "-")
191 language = locale.split("-")[0]
192 headers["Accept-Language"] = f"{locale}, {language};q=0.9, *;q=0.5"
193
194 # Prepare Params
195 params = kwargs.pop("params", {}) or {}
196 # sessionId is an unofficial-API concept; don't send it to the official API.
197 if self.auth.session_id and base_url != OPEN_API_URL:
198 params["sessionId"] = self.auth.session_id
199 if self.auth.country_code:
200 params["countryCode"] = self.auth.country_code
201
202 self.logger.debug("Making %s request to Tidal API: %s", method, endpoint)
203
204 async with self.mass.http_session.request(
205 method, url, headers=headers, params=params, **kwargs
206 ) as response:
207 if response.status != 401:
208 return await self._handle_response(response)
209
210 # The token was rejected before its known expiry (e.g. invalidated
211 # server-side): force a refresh and retry the request once.
212 self.logger.debug("Got 401 from Tidal API, forcing token refresh and retrying")
213 if not await self.auth.refresh_token():
214 raise LoginFailed("Authentication failed")
215 headers["Authorization"] = f"Bearer {self.auth.access_token}"
216
217 async with self.mass.http_session.request(
218 method, url, headers=headers, params=params, **kwargs
219 ) as response:
220 return await self._handle_response(response)
221
222 async def _handle_response(self, response: ClientResponse) -> tuple[dict[str, Any], str]:
223 """Handle API response and common error conditions."""
224 if response.status == 401:
225 raise LoginFailed("Authentication failed")
226 if response.status == 404:
227 raise MediaNotFoundError(f"Item not found: {response.url}")
228 if response.status == 429:
229 retry_after = int(response.headers.get("Retry-After", 30))
230 raise RateLimited("Tidal Rate limit reached", backoff_time=retry_after)
231 if response.status >= 400:
232 text = await response.text()
233 self.logger.error("API error: %s - %s", response.status, text)
234 raise ResourceTemporarilyUnavailable("API error")
235
236 try:
237 if response.status == 204 or response.content_length == 0:
238 data = {"success": True}
239 else:
240 data = await response.json()
241 except json.JSONDecodeError as err:
242 raise ResourceTemporarilyUnavailable("Failed to parse response") from err
243
244 return data, response.headers.get("ETag", "")
245