/
/
/
1"""Helpers for consuming JSON:API responses from the official Tidal API."""
2
3from __future__ import annotations
4
5from typing import Any
6from urllib.parse import parse_qs, urlparse
7
8
9class JsonApiDocument:
10 """
11 Wrapper around a parsed JSON:API document.
12
13 Indexes the ``included`` resources by (type, id) so that relationship
14 linkages on the primary resource(s) can be resolved to the full resource
15 objects returned via ``?include=``.
16 """
17
18 def __init__(self, raw: dict[str, Any]) -> None:
19 """Initialize the document from a parsed JSON:API response."""
20 self.raw = raw
21 self._included: dict[tuple[str, str], dict[str, Any]] = {
22 (resource["type"], resource["id"]): resource
23 for resource in raw.get("included", [])
24 if resource.get("type") and resource.get("id")
25 }
26
27 @property
28 def data(self) -> dict[str, Any]:
29 """Return the primary single-resource object."""
30 data = self.raw.get("data")
31 return data if isinstance(data, dict) else {}
32
33 @property
34 def data_list(self) -> list[dict[str, Any]]:
35 """Return the primary resource collection."""
36 data = self.raw.get("data")
37 return data if isinstance(data, list) else []
38
39 @property
40 def next_cursor(self) -> str | None:
41 """Return the opaque cursor for the next page, if any."""
42 links = self.raw.get("links") or {}
43 next_link = links.get("next")
44 if not isinstance(next_link, str) or not next_link:
45 return None
46 # The next link is a path with a page[cursor] query param, whose brackets
47 # may be percent-encoded (page%5Bcursor%5D). parse_qs decodes both the key
48 # and value. A next link WITHOUT that param is not a usable cursor: guessing
49 # (e.g. sending the whole link) would fire a malformed request, so stop the
50 # walk cleanly instead.
51 if cursors := parse_qs(urlparse(next_link).query).get("page[cursor]"):
52 return cursors[0]
53 return None
54
55 def resolve(self, identifier: dict[str, Any]) -> dict[str, Any] | None:
56 """Resolve a resource identifier ({type, id}) to its included resource."""
57 if "type" not in identifier or "id" not in identifier:
58 return None
59 return self._included.get((identifier["type"], identifier["id"]))
60
61 def linkage_ids(self, resource: dict[str, Any], relationship: str) -> set[str]:
62 """Return the ids of a resource's relationship linkage (no include needed)."""
63 rel = (resource.get("relationships") or {}).get(relationship) or {}
64 linkage = rel.get("data")
65 if linkage is None:
66 return set()
67 if isinstance(linkage, dict):
68 linkage = [linkage]
69 return {str(item["id"]) for item in linkage if item.get("id")}
70
71 def related(self, resource: dict[str, Any], relationship: str) -> list[dict[str, Any]]:
72 """
73 Resolve a resource's relationship to the included resource objects.
74
75 :param resource: The resource object whose relationship to resolve.
76 :param relationship: The relationship name (e.g. "artists", "coverArt").
77 """
78 rel = (resource.get("relationships") or {}).get(relationship) or {}
79 linkage = rel.get("data")
80 if linkage is None:
81 return []
82 if isinstance(linkage, dict):
83 linkage = [linkage]
84 resolved = []
85 for identifier in linkage:
86 key = (identifier.get("type"), identifier.get("id"))
87 if included := self._included.get(key):
88 resolved.append(included)
89 return resolved
90
91 def related_one(self, resource: dict[str, Any], relationship: str) -> dict[str, Any] | None:
92 """Resolve a to-one relationship to a single included resource object."""
93 resolved = self.related(resource, relationship)
94 return resolved[0] if resolved else None
95