music-assistant-server

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