From b5efa1bc3285838c2835408f781aaf8983f9e772 Mon Sep 17 00:00:00 2001 From: Aru Sahni Date: Sun, 27 Aug 2023 22:40:41 -0400 Subject: [PATCH] Add typing to the Page classes --- tidalapi/page.py | 221 +++++++++++++++++++++++++++----------------- tidalapi/request.py | 37 +++++++- tidalapi/session.py | 37 +++++--- 3 files changed, 192 insertions(+), 103 deletions(-) diff --git a/tidalapi/page.py b/tidalapi/page.py index e3a66f5..8162194 100644 --- a/tidalapi/page.py +++ b/tidalapi/page.py @@ -19,15 +19,49 @@ """ import copy -from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Union, cast +from dataclasses import dataclass +from typing import ( + TYPE_CHECKING, + Any, + Callable, + Dict, + Iterator, + List, + Optional, + Union, + cast, +) from tidalapi.types import JsonObj if TYPE_CHECKING: - import tidalapi + from tidalapi.album import Album + from tidalapi.artist import Artist + from tidalapi.media import Track, Video + from tidalapi.mix import Mix + from tidalapi.playlist import Playlist, UserPlaylist + from tidalapi.request import Requests + from tidalapi.session import Session +PageCategories = Union[ + "Album", + "PageLinks", + "FeaturedItems", + "ItemList", + "TextBlock", + "LinkList", + "Mix", +] -class Page(object): +AllCategories = Union["Artist", PageCategories] + + +def _no_op_items(*args) -> List[Any]: + """A simple passthrough stub for category types that are not iterable.""" + return [] + + +class Page: """ A page from the https://listen.tidal.com/view/pages/ endpoint @@ -35,26 +69,33 @@ class Page(object): However it is an iterable that goes through all the visible items on the page as well, in the natural reading order """ - title = "" - categories: Optional[List[Any]] = None - _categories_iter: Optional[Iterator[Any]] = None + title: Optional[str] = "" + categories: Optional[List["AllCategories"]] = None + _categories_iter: Optional[Iterator["AllCategories"]] = None + _items_iter: Optional[Iterator[Callable[..., Any]]] = None + page_category: "PageCategory" + request: "Requests" - def __init__(self, session, title): + def __init__(self, session: "Session", title: Optional[str]): self.request = session.request self.categories = None self.title = title self.page_category = PageCategory(session) - def __iter__(self): + def __iter__(self) -> "Page": if self.categories is None: raise AttributeError("No categories found") self._categories_iter = iter(self.categories) self._category = next(self._categories_iter) - self._items_iter = iter(self._category.items) + self._items_iter = iter( + cast( + List[Callable[..., Any]], getattr(self._category, "items", _no_op_items) + ) + ) return self - def __next__(self): - if self._category == StopIteration: + def __next__(self) -> Callable[..., Any]: + if self._items_iter is None: return StopIteration try: item = next(self._items_iter) @@ -62,11 +103,16 @@ def __next__(self): if self._categories_iter is None: raise AttributeError("No categories found") self._category = next(self._categories_iter) - self._items_iter = iter(self._category.items) + self._items_iter = iter( + cast( + List[Callable[..., Any]], + getattr(self._category, "items", _no_op_items), + ) + ) return self.__next__() return item - def next(self): + def next(self) -> Callable[..., Any]: return self.__next__() def parse(self, json_obj: JsonObj) -> "Page": @@ -99,17 +145,30 @@ def get(self, endpoint: str, params: Optional[Dict[str, Any]] = None) -> "Page": return self.parse(json_obj) -class PageCategory(object): +@dataclass +class More: + api_path: str + title: str + + @staticmethod + def from_response(json_obj: JsonObj) -> Optional["More"]: + show_more = json_obj.get("showMore") + if show_more is None: + return None + return More(api_path=show_more["apiPath"], title=show_more["title"]) + + +class PageCategory: type = None - title = None + title: Optional[str] = None description: Optional[str] = "" - requests = None - _more: Optional[dict[str, Union[dict[str, str], str]]] = None + request: "Requests" + _more: Optional[More] = None - def __init__(self, session: "tidalapi.session.Session"): + def __init__(self, session: "Session"): self.session = session self.request = session.request - self.item_types = { + self.item_types: Dict[str, Callable[..., Any]] = { "ALBUM_LIST": self.session.parse_album, "ARTIST_LIST": self.session.parse_artist, "TRACK_LIST": self.session.parse_track, @@ -118,25 +177,23 @@ def __init__(self, session: "tidalapi.session.Session"): "MIX_LIST": self.session.parse_mix, } - def parse(self, json_obj): + def parse(self, json_obj: JsonObj) -> AllCategories: result = None category_type = json_obj["type"] if category_type in ("PAGE_LINKS_CLOUD", "PAGE_LINKS"): - category: Union[ - PageLinks, FeaturedItems, ItemList, TextBlock, LinkList - ] = PageLinks(self.session) + category: PageCategories = PageLinks(self.session) elif category_type in ("FEATURED_PROMOTIONS", "MULTIPLE_TOP_PROMOTIONS"): category = FeaturedItems(self.session) elif category_type in self.item_types.keys(): category = ItemList(self.session) elif category_type == "MIX_HEADER": - return self.session.parse_mix(json_obj["mix"]) + return cast("Mix", self.session.parse_mix(json_obj["mix"])) elif category_type == "ARTIST_HEADER": - result = self.session.parse_artist(json_obj["artist"]) + result = cast("Artist", self.session.parse_artist(json_obj["artist"])) result.bio = json_obj["bio"] return result elif category_type == "ALBUM_HEADER": - return self.session.parse_album(json_obj["album"]) + return cast("Album", self.session.parse_album(json_obj["album"])) elif category_type == "HIGHLIGHT_MODULE": category = ItemList(self.session) elif category_type == "MIXED_TYPES_LIST": @@ -152,25 +209,19 @@ def parse(self, json_obj): json_obj["items"] = json_obj["socialProfiles"] category = LinkList(self.session) else: - raise NotImplementedError( - "PageType {} not implemented".format(category_type) - ) + raise NotImplementedError(f"PageType {category_type} not implemented") return category.parse(json_obj) - def show_more(self): + def show_more(self) -> Optional[Page]: """Get the full list of items on their own :class:`.Page` from a :class:`.PageCategory` :return: A :class:`.Page` more of the items in the category, None if there aren't any """ - if self._more: - api_path = self._more["apiPath"] - assert isinstance(api_path, str) - else: - api_path = None + api_path = self._more.api_path if self._more else None return ( - Page(self.session, self._more["title"]).get(api_path) + Page(self.session, self._more.title).get(api_path) if api_path and self._more else None ) @@ -179,12 +230,12 @@ def show_more(self): class FeaturedItems(PageCategory): """Items that have been featured by TIDAL.""" - items: Optional[list["PageItem"]] = None + items: Optional[List["PageItem"]] = None - def __init__(self, session): + def __init__(self, session: "Session"): super(FeaturedItems, self).__init__(session) - def parse(self, json_obj): + def parse(self, json_obj: JsonObj) -> "FeaturedItems": self.items = [] self.title = json_obj["title"] self.description = json_obj["description"] @@ -198,15 +249,15 @@ def parse(self, json_obj): class PageLinks(PageCategory): """A list of :class:`.PageLink` to other parts of TIDAL.""" - items: Optional[list["PageLink"]] = None + items: Optional[List["PageLink"]] = None - def parse(self, json_obj): + def parse(self, json_obj: JsonObj) -> "PageLinks": """Parse the list of links from TIDAL. :param json_obj: The json to be parsed :return: A copy of this page category containing the links in the items field """ - self._more = json_obj.get("showMore") + self._more = More.from_response(json_obj) self.title = json_obj["title"] self.items = [] for item in json_obj["pagedList"]["items"]: @@ -219,20 +270,20 @@ class ItemList(PageCategory): """A list of items from TIDAL, can be a list of mixes, for example, or a list of playlists and mixes in some cases.""" - items = None + items: Optional[List[Any]] = None - def parse(self, json_obj): + def parse(self, json_obj: JsonObj) -> "ItemList": """Parse a list of items on TIDAL from the pages endpoints. :param json_obj: The json from TIDAL to be parsed :return: A copy of the ItemList with a list of items """ - self._more = json_obj.get("showMore") + self._more = More.from_response(json_obj) self.title = json_obj["title"] item_type = json_obj["type"] list_key = "pagedList" - session = None - parse = None + session: Optional["Session"] = None + parse: Optional[Callable[..., Any]] = None if item_type in self.item_types.keys(): parse = self.item_types[item_type] @@ -254,15 +305,14 @@ def parse(self, json_obj): return copy.copy(self) -class PageLink(object): +class PageLink: """A Link to another :class:`.Page` on TIDAL, Call get() to retrieve the Page.""" - title = None + title: str icon = None image_id = None - requests = None - def __init__(self, session: "tidalapi.session.Session", json_obj): + def __init__(self, session: "Session", json_obj: JsonObj): self.session = session self.request = session.request self.title = json_obj["title"] @@ -270,30 +320,34 @@ def __init__(self, session: "tidalapi.session.Session", json_obj): self.api_path = cast(str, json_obj["apiPath"]) self.image_id = json_obj["imageId"] - def get(self): + def get(self) -> "Page": """Requests the linked page from TIDAL :return: A :class:`Page` at the api_path.""" - return self.request.map_request( - self.api_path, - params={"deviceType": "DESKTOP"}, - parse=self.session.parse_page, + return cast( + "Page", + self.request.map_request( + self.api_path, + params={"deviceType": "DESKTOP"}, + parse=self.session.parse_page, + ), ) -class PageItem(object): +class PageItem: """An Item from a :class:`.PageCategory` from the /pages endpoint, call get() to retrieve the actual item.""" - header = "" - short_header = "" - short_sub_header = "" - image_id = "" - type = "" - artifact_id = "" - text = "" - featured = False - - def __init__(self, session, json_obj): + header: str = "" + short_header: str = "" + short_sub_header: str = "" + image_id: str = "" + type: str = "" + artifact_id: str = "" + text: str = "" + featured: bool = False + session: "Session" + + def __init__(self, session: "Session", json_obj: JsonObj): self.session = session self.request = session.request self.header = json_obj["header"] @@ -305,37 +359,34 @@ def __init__(self, session, json_obj): self.text = json_obj["text"] self.featured = bool(json_obj["featured"]) - def get(self): + def get(self) -> Union["Artist", "Playlist", "Track", "UserPlaylist", "Video"]: """Retrieve the PageItem with the artifact_id matching the type. :return: The fully parsed item, e.g. :class:`.Playlist`, :class:`.Video`, :class:`.Track` """ if self.type == "PLAYLIST": - result = self.session.playlist(self.artifact_id) + return self.session.playlist(self.artifact_id) elif self.type == "VIDEO": - result = self.session.video(self.artifact_id) + return self.session.video(self.artifact_id) elif self.type == "TRACK": - result = self.session.track(self.artifact_id) + return self.session.track(self.artifact_id) elif self.type == "ARTIST": - result = self.session.artist(self.artifact_id) - else: - raise NotImplementedError("PageItem type %s not implemented" % self.type) - - return result + return self.session.artist(self.artifact_id) + raise NotImplementedError(f"PageItem type {self.type} not implemented") class TextBlock(object): """A block of text, with a named icon, which seems to be left up to the application.""" - text = "" - icon = "" - items = None + text: str = "" + icon: str = "" + items: Optional[List[str]] = None - def __init__(self, session): + def __init__(self, session: "Session"): self.session = session - def parse(self, json_obj): + def parse(self, json_obj: JsonObj) -> "TextBlock": self.text = json_obj["text"] self.icon = json_obj["icon"] self.items = [self.text] @@ -346,11 +397,11 @@ def parse(self, json_obj): class LinkList(PageCategory): """A list of items containing links, e.g. social links or articles.""" - items = None - title = None - description = None + items: Optional[List[Any]] = None + title: Optional[str] = None + description: Optional[str] = None - def parse(self, json_obj): + def parse(self, json_obj: JsonObj) -> "LinkList": self.items = json_obj["items"] self.title = json_obj["title"] self.description = json_obj["description"] diff --git a/tidalapi/request.py b/tidalapi/request.py index ec5c637..7c8e200 100644 --- a/tidalapi/request.py +++ b/tidalapi/request.py @@ -19,7 +19,17 @@ import json import logging -from typing import Any, Callable, List, Literal, Mapping, Optional, Union +from typing import ( + TYPE_CHECKING, + Any, + Callable, + List, + Literal, + Mapping, + Optional, + Union, + cast, +) from urllib.parse import urljoin from tidalapi.types import JsonObj @@ -28,6 +38,9 @@ Params = Mapping[str, Union[str, int, None]] +if TYPE_CHECKING: + from tidalapi.session import Session + class Requests(object): """A class for handling api requests to TIDAL.""" @@ -131,10 +144,17 @@ def map_request( return self.map_json(json_obj, parse=parse) @classmethod - def map_json(cls, json_obj, parse=None, session=None): + def map_json( + cls, + json_obj: JsonObj, + parse: Optional[Callable] = None, + session: Optional["Session"] = None, + ) -> List[Any]: items = json_obj.get("items") if items is None: + if parse is None: + raise ValueError("A parser must be supplied") return parse(json_obj) if len(items) > 0 and "item" in items[0]: @@ -143,15 +163,22 @@ def map_json(cls, json_obj, parse=None, session=None): for item in items: item["item"]["dateAdded"] = item["created"] - lists = [] + lists: List[Any] = [] for item in items: if session is not None: - parse = session.convert_type( - item["type"].lower() + "s", output="parse" + parse = cast( + Callable, + session.convert_type( + cast(str, item["type"]).lower() + "s", output="parse" + ), ) + if parse is None: + raise ValueError("A parser must be supplied") lists.append(parse(item["item"])) return lists + if parse is None: + raise ValueError("A parser must be supplied") return list(map(parse, items)) def get_items(self, url, parse): diff --git a/tidalapi/session.py b/tidalapi/session.py index b725997..f03a439 100644 --- a/tidalapi/session.py +++ b/tidalapi/session.py @@ -34,6 +34,7 @@ List, Literal, Optional, + TypedDict, Union, cast, no_type_check, @@ -192,6 +193,15 @@ class TypeRelation: parse: Callable +class SearchResults(TypedDict): + artists: List[artist.Artist] + albums: List[album.Album] + tracks: List[media.Track] + videos: List[media.Video] + playlists: List[Union[playlist.Playlist, playlist.UserPlaylist]] + top_hit: Optional[List[Any]] + + class Session(object): """Object for interacting with the TIDAL api and.""" @@ -258,12 +268,12 @@ def __init__(self, config=Config()): def convert_type( self, - search, + search: str, search_type: TypeConversionKeys = "identifier", output: TypeConversionKeys = "identifier", - case=Case.lower, - suffix=True, - ): + case: Case = Case.lower, + suffix: bool = True, + ) -> Union[str, Callable]: type_relations = next( x for x in self.type_conversions if getattr(x, search_type) == search ) @@ -483,7 +493,7 @@ def video_quality(self) -> str: def video_quality(self, quality): self.config.video_quality = media.VideoQuality(quality).value - def search(self, query, models=None, limit=50, offset=0): + def search(self, query, models=None, limit=50, offset=0) -> SearchResults: """Searches TIDAL with the specified query, you can also specify what models you want to search for. While you can set the offset, there aren't more than 300 items available in a search. @@ -504,7 +514,7 @@ def search(self, query, models=None, limit=50, offset=0): for model in models: if model not in SearchTypes: raise ValueError("Tried to search for an invalid type") - types.append(self.convert_type(model, "type")) + types.append(cast(str, self.convert_type(model, "type"))) params = { "query": query, @@ -515,7 +525,7 @@ def search(self, query, models=None, limit=50, offset=0): json_obj = self.request.request("GET", "search", params=params).json() - result = { + result: SearchResults = { "artists": self.request.map_json(json_obj["artists"], self.parse_artist), "albums": self.request.map_json(json_obj["albums"], self.parse_album), "tracks": self.request.map_json(json_obj["tracks"], self.parse_track), @@ -523,6 +533,7 @@ def search(self, query, models=None, limit=50, offset=0): "playlists": self.request.map_json( json_obj["playlists"], self.parse_playlist ), + "top_hit": None, } # Find the type of the top hit so we can parse it @@ -530,10 +541,8 @@ def search(self, query, models=None, limit=50, offset=0): top_type = json_obj["topHit"]["type"].lower() parse = self.convert_type(top_type, output="parse") result["top_hit"] = self.request.map_json( - json_obj["topHit"]["value"], parse + json_obj["topHit"]["value"], cast(Callable[..., Any], parse) ) - else: - result["top_hit"] = None return result @@ -546,7 +555,7 @@ def check_login(self): ).ok def playlist( - self, playlist_id=None + self, playlist_id: Optional[str] = None ) -> Union[tidalapi.Playlist, tidalapi.UserPlaylist]: """Function to create a playlist object with access to the session instance in a smoother way. Calls :class:`tidalapi.Playlist(session=session, @@ -558,7 +567,9 @@ def playlist( return playlist.Playlist(session=self, playlist_id=playlist_id).factory() - def track(self, track_id=None, with_album=False) -> tidalapi.Track: + def track( + self, track_id: Optional[str] = None, with_album: bool = False + ) -> tidalapi.Track: """Function to create a Track object with access to the session instance in a smoother way. Calls :class:`tidalapi.Track(session=session, track_id=track_id) <.Track>` internally. @@ -576,7 +587,7 @@ def track(self, track_id=None, with_album=False) -> tidalapi.Track: return item - def video(self, video_id=None) -> tidalapi.Video: + def video(self, video_id: Optional[str] = None) -> tidalapi.Video: """Function to create a Video object with access to the session instance in a smoother way. Calls :class:`tidalapi.Video(session=session, video_id=video_id) <.Video>` internally.