/
/
/
1"""API Client for Tidal."""
2
3from __future__ import annotations
4
5import json
6from typing import TYPE_CHECKING, Any, cast
7
8from music_assistant_models.errors import (
9 LoginFailed,
10 MediaNotFoundError,
11 ResourceTemporarilyUnavailable,
12)
13
14from music_assistant.helpers.throttle_retry import ThrottlerManager, throttle_with_retries
15
16if TYPE_CHECKING:
17 from collections.abc import AsyncGenerator
18
19 from aiohttp import ClientResponse
20
21 from .provider import TidalProvider
22
23
24class TidalAPIClient:
25 """Client for interacting with Tidal API."""
26
27 BASE_URL: str = "https://api.tidal.com/v1"
28 BASE_URL_V2: str = "https://api.tidal.com/v2"
29 OPEN_API_URL: str = "https://openapi.tidal.com/v2"
30
31 # Define throttler here for use by the client
32 throttler = ThrottlerManager(rate_limit=1, period=2)
33
34 def __init__(self, provider: TidalProvider):
35 """Initialize API client."""
36 self.provider = provider
37 self.auth = provider.auth
38 self.logger = provider.logger
39 self.mass = provider.mass
40
41 async def get(
42 self, endpoint: str, **kwargs: Any
43 ) -> dict[str, Any] | tuple[dict[str, Any], str]:
44 """Get data from Tidal API."""
45 return await self._request("GET", endpoint, **kwargs)
46
47 async def get_data(self, endpoint: str, **kwargs: Any) -> dict[str, Any]:
48 """Get data from Tidal API, discarding headers/ETags."""
49 result = await self.get(endpoint, **kwargs)
50 return result[0] if isinstance(result, tuple) else result
51
52 async def post(
53 self,
54 endpoint: str,
55 data: dict[str, Any] | None = None,
56 as_form: bool = False,
57 **kwargs: Any,
58 ) -> dict[str, Any]:
59 """Send POST data to Tidal API."""
60 if as_form:
61 kwargs.setdefault("headers", {})["Content-Type"] = "application/x-www-form-urlencoded"
62 kwargs["data"] = data
63 else:
64 kwargs["json"] = data
65
66 return cast("dict[str, Any]", await self._request("POST", endpoint, **kwargs))
67
68 async def put(
69 self,
70 endpoint: str,
71 data: dict[str, Any] | None = None,
72 as_form: bool = False,
73 **kwargs: Any,
74 ) -> dict[str, Any]:
75 """Send PUT data to Tidal API."""
76 # Special handling for mixes which use V2
77 if "mixes" in endpoint and "base_url" not in kwargs:
78 kwargs["base_url"] = self.BASE_URL_V2
79
80 if as_form:
81 kwargs.setdefault("headers", {})["Content-Type"] = "application/x-www-form-urlencoded"
82 kwargs["data"] = data
83 else:
84 kwargs["json"] = data
85
86 return cast("dict[str, Any]", await self._request("PUT", endpoint, **kwargs))
87
88 async def delete(
89 self, endpoint: str, data: dict[str, Any] | None = None, **kwargs: Any
90 ) -> dict[str, Any]:
91 """Delete data from Tidal API."""
92 kwargs["json"] = data
93 return cast("dict[str, Any]", await self._request("DELETE", endpoint, **kwargs))
94
95 @throttle_with_retries # type: ignore[type-var]
96 async def _request(
97 self, method: str, endpoint: str, **kwargs: Any
98 ) -> dict[str, Any] | tuple[dict[str, Any], str]:
99 """Handle API requests internally."""
100 if not await self.auth.ensure_valid_token():
101 raise LoginFailed("Failed to authenticate with Tidal")
102
103 # Prepare URL
104 base_url = kwargs.pop("base_url", self.BASE_URL)
105 url = f"{base_url}/{endpoint}"
106
107 # Prepare Headers
108 headers = kwargs.pop("headers", {})
109 headers["Authorization"] = f"Bearer {self.auth.access_token}"
110
111 locale = self.mass.metadata.locale.replace("_", "-")
112 language = locale.split("-")[0]
113 headers["Accept-Language"] = f"{locale}, {language};q=0.9, *;q=0.5"
114
115 # Prepare Params
116 params = kwargs.pop("params", {}) or {}
117 if self.auth.session_id:
118 params["sessionId"] = self.auth.session_id
119 if self.auth.country_code:
120 params["countryCode"] = self.auth.country_code
121
122 # Extract special handling flags
123 return_etag = kwargs.pop("return_etag", False)
124
125 self.logger.debug("Making %s request to Tidal API: %s", method, endpoint)
126
127 async with self.mass.http_session.request(
128 method, url, headers=headers, params=params, **kwargs
129 ) as response:
130 return await self._handle_response(response, return_etag)
131
132 async def _handle_response(
133 self, response: ClientResponse, return_etag: bool = False
134 ) -> dict[str, Any] | tuple[dict[str, Any], str]:
135 """Handle API response and common error conditions."""
136 if response.status == 401:
137 raise LoginFailed("Authentication failed")
138 if response.status == 404:
139 raise MediaNotFoundError(f"Item not found: {response.url}")
140 if response.status == 429:
141 retry_after = int(response.headers.get("Retry-After", 30))
142 raise ResourceTemporarilyUnavailable(
143 "Tidal Rate limit reached", backoff_time=retry_after
144 )
145 if response.status >= 400:
146 text = await response.text()
147 self.logger.error("API error: %s - %s", response.status, text)
148 raise ResourceTemporarilyUnavailable("API error")
149
150 try:
151 if response.status == 204 or response.content_length == 0:
152 data = {"success": True}
153 else:
154 data = await response.json()
155
156 if return_etag:
157 etag = response.headers.get("ETag", "")
158 return data, etag
159 return data
160 except json.JSONDecodeError as err:
161 raise ResourceTemporarilyUnavailable("Failed to parse response") from err
162
163 async def paginate(
164 self,
165 endpoint: str,
166 item_key: str = "items",
167 limit: int = 50,
168 cursor_based: bool = False,
169 **kwargs: Any,
170 ) -> AsyncGenerator[Any, None]:
171 """Paginate through all items from a Tidal API endpoint."""
172 offset = 0
173 cursor = None
174
175 while True:
176 params = {"limit": limit}
177 if cursor_based:
178 if cursor:
179 params["cursor"] = cursor
180 else:
181 params["offset"] = offset
182
183 if "params" in kwargs:
184 params.update(kwargs.pop("params"))
185
186 api_result = await self.get(endpoint, params=params, **kwargs)
187 response = api_result[0] if isinstance(api_result, tuple) else api_result
188
189 items = response.get(item_key, [])
190 if not items:
191 break
192
193 for item in items:
194 yield item
195
196 if cursor_based:
197 cursor = response.get("cursor")
198 if not cursor:
199 break
200 else:
201 offset += len(items)
202