From c4545549795eea94d7b4b7a4cae5d71b5d6db40f 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/artist.py | 8 ++ tidalapi/mix.py | 2 +- tidalapi/page.py | 210 ++++++++++++++++++++++++++------------------ tidalapi/request.py | 37 ++++++-- tidalapi/session.py | 68 ++++++++------ 5 files changed, 209 insertions(+), 116 deletions(-) diff --git a/tidalapi/artist.py b/tidalapi/artist.py index 7c3fb2a..e7fdf31 100644 --- a/tidalapi/artist.py +++ b/tidalapi/artist.py @@ -196,6 +196,14 @@ def get_radio(self) -> List["Track"]: ), ) + def items(self) -> list: + """The artist page does not supply any items. This only exists for symmetry with + other model types. + + :return: An empty list. + """ + return [] + def image(self, dimensions: int = 320) -> str: """A url to an artist picture. diff --git a/tidalapi/mix.py b/tidalapi/mix.py index 96d16a0..f6d4620 100644 --- a/tidalapi/mix.py +++ b/tidalapi/mix.py @@ -72,7 +72,7 @@ class Mix: _retrieved = False _items: Optional[List[Union["Video", "Track"]]] = None - def __init__(self, session: Session, mix_id: str): + def __init__(self, session: Session, mix_id: Optional[str]): self.session = session self.request = session.request if mix_id is not None: diff --git a/tidalapi/page.py b/tidalapi/page.py index e3a66f5..1ba0f68 100644 --- a/tidalapi/page.py +++ b/tidalapi/page.py @@ -19,15 +19,44 @@ """ 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 - - -class Page(object): + 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", +] + +AllCategories = Union["Artist", PageCategories] + + +class Page: """ A page from the https://listen.tidal.com/view/pages/ endpoint @@ -35,26 +64,29 @@ 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: 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: 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]], self._category.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 +94,13 @@ 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]], self._category.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 +133,31 @@ 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 + + @classmethod + def parse(cls, json_obj: JsonObj) -> Optional["More"]: + show_more = json_obj.get("showMore") + if show_more is None: + return None + else: + return cls(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,13 +166,11 @@ 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(): @@ -152,25 +198,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 +219,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): - super(FeaturedItems, self).__init__(session) + def __init__(self, session: "Session"): + super().__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 +238,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.parse(json_obj) self.title = json_obj["title"] self.items = [] for item in json_obj["pagedList"]["items"]: @@ -219,20 +259,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.parse(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 +294,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 +309,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 +348,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 +386,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..910a49c 100644 --- a/tidalapi/session.py +++ b/tidalapi/session.py @@ -28,12 +28,12 @@ from dataclasses import dataclass from enum import Enum from typing import ( - TYPE_CHECKING, Any, Callable, List, Literal, Optional, + TypedDict, Union, cast, no_type_check, @@ -42,8 +42,7 @@ import requests -if TYPE_CHECKING: - import tidalapi +from tidalapi.types import JsonObj from . import album, artist, genre, media, mix, page, playlist, request, user @@ -192,6 +191,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.""" @@ -219,18 +227,15 @@ def __init__(self, config=Config()): self.request = request.Requests(session=self) self.genre = genre.Genre(session=self) - self.parse_album = self.album().parse - self.parse_artist = self.artist().parse_artist self.parse_artists = self.artist().parse_artists self.parse_playlist = self.playlist().parse self.parse_track = self.track().parse_track self.parse_video = self.video().parse_video self.parse_media = self.track().parse_media - self.parse_mix = self.mix().parse self.parse_user = user.User(self, None).parse - self.page = page.Page(self, None) + self.page = page.Page(self, "") self.parse_page = self.page.parse self.type_conversions: List[TypeRelation] = [ @@ -256,14 +261,26 @@ def __init__(self, config=Config()): ) ] + def parse_album(self, obj: JsonObj) -> album.Album: + """Parse an album from the given response.""" + return self.album().parse(obj) + + def parse_artist(self, obj: JsonObj) -> artist.Artist: + """Parse an artist from the given response.""" + return self.artist().parse_artist(obj) + + def parse_mix(self, obj: JsonObj) -> mix.Mix: + """Parse a mix from the given response.""" + return self.mix().parse(obj) + 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 +500,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 +521,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 +532,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 +540,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 +548,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,8 +562,8 @@ def check_login(self): ).ok def playlist( - self, playlist_id=None - ) -> Union[tidalapi.Playlist, tidalapi.UserPlaylist]: + self, playlist_id: Optional[str] = None + ) -> Union[playlist.Playlist, playlist.UserPlaylist]: """Function to create a playlist object with access to the session instance in a smoother way. Calls :class:`tidalapi.Playlist(session=session, playlist_id=playlist_id) <.Playlist>` internally. @@ -558,7 +574,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 + ) -> media.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 +594,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) -> media.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. @@ -587,7 +605,7 @@ def video(self, video_id=None) -> tidalapi.Video: return media.Video(session=self, media_id=video_id) - def artist(self, artist_id: Optional[str] = None) -> tidalapi.Artist: + def artist(self, artist_id: Optional[str] = None) -> artist.Artist: """Function to create a Artist object with access to the session instance in a smoother way. Calls :class:`tidalapi.Artist(session=session, artist_id=artist_id) <.Artist>` internally. @@ -598,7 +616,7 @@ def artist(self, artist_id: Optional[str] = None) -> tidalapi.Artist: return artist.Artist(session=self, artist_id=artist_id) - def album(self, album_id: Optional[str] = None) -> tidalapi.Album: + def album(self, album_id: Optional[str] = None) -> album.Album: """Function to create a Album object with access to the session instance in a smoother way. Calls :class:`tidalapi.Album(session=session, album_id=album_id) <.Album>` internally. @@ -609,7 +627,7 @@ def album(self, album_id: Optional[str] = None) -> tidalapi.Album: return album.Album(session=self, album_id=album_id) - def mix(self, mix_id=None) -> tidalapi.Mix: + def mix(self, mix_id: Optional[str] = None) -> mix.Mix: """Function to create a mix object with access to the session instance smoothly Calls :class:`tidalapi.Mix(session=session, mix_id=mix_id) <.Album>` internally. @@ -621,7 +639,7 @@ def mix(self, mix_id=None) -> tidalapi.Mix: def get_user( self, user_id=None - ) -> Union[tidalapi.FetchedUser, tidalapi.LoggedInUser, tidalapi.PlaylistCreator]: + ) -> Union[user.FetchedUser, user.LoggedInUser, user.PlaylistCreator]: """Function to create a User object with access to the session instance in a smoother way. Calls :class:`user.User(session=session, user_id=user_id) <.User>` internally.