/
/
/
1"""API Client for YouSee Musik."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any
6
7from music_assistant_models.errors import (
8 LoginFailed,
9)
10
11from music_assistant.constants import VERBOSE_LOG_LEVEL
12from music_assistant.helpers.json import json_dumps
13from music_assistant.helpers.throttle_retry import ThrottlerManager, throttle_with_retries
14from music_assistant.providers.yousee.constants import MAX_PAGES_PAGINATED, PAGE_SIZE
15
16if TYPE_CHECKING:
17 from collections.abc import AsyncGenerator
18
19 from music_assistant.providers.yousee.provider import YouSeeMusikProvider
20
21
22JsonLike = dict[str, Any]
23
24
25class YouSeeGraphQLError(Exception):
26 """YouSee Musik GraphQL error."""
27
28 def __init__(self, data: JsonLike) -> None:
29 """Initialize YouSeeGraphQLError."""
30 super().__init__(json_dumps(data))
31
32
33class YouSeeAPIClient:
34 """Client for interacting with YouSee API."""
35
36 YOUSEE_GRAPHQL_ENDPOINT = "https://graphql-1458.api.247e.com/graphql"
37
38 # Unsure if yousee enforces rate limiting, this is just a sane precaution
39 throttler = ThrottlerManager(rate_limit=4, period=1)
40
41 def __init__(self, provider: YouSeeMusikProvider):
42 """Initialize API client."""
43 self.provider = provider
44 self.auth = provider.auth
45 self.logger = provider.logger
46 self.mass = provider.mass
47
48 @throttle_with_retries # type: ignore[type-var]
49 async def post_graphql(
50 self, query: str, variables: JsonLike, _headers: JsonLike | None = None
51 ) -> JsonLike:
52 """Post GraphQL query to YouSee endpoint with authorization."""
53 locale = self.mass.metadata.locale.split("_")[0]
54
55 async with self.mass.http_session.post(
56 self.YOUSEE_GRAPHQL_ENDPOINT,
57 json={"query": query, "variables": variables},
58 headers={
59 "Authorization": f"Bearer {await self.auth.auth_token()}",
60 "Accept-Language": locale,
61 }
62 | (_headers or {}),
63 ) as resp:
64 if resp.status in {401, 403}:
65 # Invalidate token
66 self.auth.invalidate()
67 raise LoginFailed("Authentication with YouSee failed")
68
69 resp.raise_for_status()
70
71 result = await resp.json()
72 if len(result.get("errors", [])) > 0:
73 raise YouSeeGraphQLError(result)
74
75 return dict(result)
76
77 async def paginate_graphql(
78 self,
79 query: str,
80 variables: JsonLike,
81 page_path: list[str],
82 variables_first_key: str = "first",
83 variables_after_key: str = "after",
84 ) -> AsyncGenerator[JsonLike, None]:
85 """Paginate GraphQL results."""
86 after = None
87 has_more = True
88 i = 0
89 while has_more and (i < MAX_PAGES_PAGINATED):
90 self.logger.log(VERBOSE_LOG_LEVEL, "Paginating GraphQL query, page %s", i + 1)
91 vars_with_pagination = variables | {
92 variables_first_key: PAGE_SIZE,
93 variables_after_key: after,
94 }
95 result = await self.post_graphql(query, vars_with_pagination)
96
97 # Navigate to the page containing items and pageInfo
98 page_data = result
99 for key in page_path:
100 page_data = page_data.get(key, {})
101
102 for item in page_data.get("items", []):
103 yield item
104
105 page_info = page_data.get("pageInfo", {})
106 has_more = page_info.get("hasNextPage", False)
107 after = page_info.get("endCursor", None)
108 i += 1
109