From bee481b1a6256e163b26788c3b9112f1a3108bbf Mon Sep 17 00:00:00 2001 From: Johannes Linde Date: Wed, 14 Oct 2026 00:01:29 +0200 Subject: [PATCH 1/3] Replace deprecated datetime.utcnow() with timezone-aware UTC datetimes --- HISTORY.rst | 1 + tests/test_token_refresh.py | 27 +++++++++++++++++++++++---- tidalapi/request.py | 2 +- tidalapi/session.py | 12 ++++++------ 4 files changed, 31 insertions(+), 11 deletions(-) diff --git a/HISTORY.rst b/HISTORY.rst index 70399b7..e40db50 100644 --- a/HISTORY.rst +++ b/HISTORY.rst @@ -16,6 +16,7 @@ v0.9.0 * Bugfix: `login_oauth` works after `client_enable_hires` has been called - tehkillerbee_ * Bugfix: Sessions created without a config no longer share one `Config` object - tehkillerbee_ * `save_session_to_file` and `load_session_from_file` now persist `expiry_time` - tehkillerbee_ +* `Session.expiry_time` is now a timezone-aware UTC datetime, replacing the deprecated `datetime.utcnow()`. Naive datetimes passed to `load_oauth_session` are still supported - tehkillerbee_ * Added `AbuseDetected`, raised when TIDAL answers with an anti-abuse HTTP 403 - tehkillerbee_ * Added `Config(backoff=...)`, enabled by default: stop sending requests after a rate limit or abuse response and raise `TidalBackoffError` until it has passed. Pass `backoff=False` to disable - tehkillerbee_ * Bugfix: Non-JSON error responses and `Retry-After` given as a date no longer break error handling - tehkillerbee_ diff --git a/tests/test_token_refresh.py b/tests/test_token_refresh.py index d857e71..27f66e4 100644 --- a/tests/test_token_refresh.py +++ b/tests/test_token_refresh.py @@ -56,6 +56,25 @@ def test_token_refresh_keeps_rotated_refresh_token(): assert session.refresh_token == "rotated-refresh" +def test_token_refresh_sets_timezone_aware_expiry_time(): + session = make_session() + session.request_session = mock.Mock() + session.request_session.post.return_value = fake_response( + json_data={ + "access_token": "new-access", + "expires_in": 3600, + "token_type": "Bearer", + } + ) + before = datetime.datetime.now(datetime.timezone.utc) + + assert session.token_refresh("old-refresh") + + assert session.expiry_time.tzinfo is datetime.timezone.utc + expected = before + datetime.timedelta(seconds=3600) + assert expected <= session.expiry_time <= expected + datetime.timedelta(seconds=5) + + def test_token_refresh_keeps_refresh_token_when_not_rotated(): session = make_session() session.request_session = mock.Mock() @@ -138,7 +157,7 @@ def _refreshing_session(expiry_time): def test_request_refreshes_expired_token_before_sending(): session = _refreshing_session( - datetime.datetime.utcnow() - datetime.timedelta(days=1) + datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=1) ) assert session.request.basic_request("GET", "tracks/1").ok @@ -150,7 +169,7 @@ def test_request_refreshes_expired_token_before_sending(): def test_request_refreshes_token_expiring_within_margin(): session = _refreshing_session( - datetime.datetime.utcnow() + datetime.timedelta(seconds=10) + datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(seconds=10) ) session.request.basic_request("GET", "tracks/1") @@ -160,7 +179,7 @@ def test_request_refreshes_token_expiring_within_margin(): def test_request_does_not_refresh_valid_token(): session = _refreshing_session( - datetime.datetime.utcnow() + datetime.timedelta(days=1) + datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=1) ) session.request.basic_request("GET", "tracks/1") @@ -187,7 +206,7 @@ def test_request_handles_timezone_aware_expiry_time(): def test_failed_proactive_refresh_does_not_fail_the_request(): session = _refreshing_session( - datetime.datetime.utcnow() + datetime.timedelta(seconds=10) + datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(seconds=10) ) session.request_session.post.return_value = fake_response( 400, {"error": "invalid_grant", "error_description": "dead"} diff --git a/tidalapi/request.py b/tidalapi/request.py index 6dee17a..a935880 100644 --- a/tidalapi/request.py +++ b/tidalapi/request.py @@ -160,7 +160,7 @@ def _access_token_expired(self) -> bool: if not expiry_time or not self.session.refresh_token: return False if expiry_time.tzinfo is None: - now = datetime.datetime.utcnow() + now = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) else: now = datetime.datetime.now(datetime.timezone.utc) return expiry_time - now <= TOKEN_EXPIRY_MARGIN diff --git a/tidalapi/session.py b/tidalapi/session.py index 2b364c0..a854fd3 100644 --- a/tidalapi/session.py +++ b/tidalapi/session.py @@ -703,9 +703,9 @@ def process_auth_token( :rtype: bool """ self.access_token = json["access_token"] - self.expiry_time = datetime.datetime.utcnow() + datetime.timedelta( - seconds=json["expires_in"] - ) + self.expiry_time = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(seconds=json["expires_in"]) self.refresh_token = json["refresh_token"] self.token_type = json["token_type"] session = self.request.request("GET", "sessions") @@ -795,9 +795,9 @@ def token_refresh(self, refresh_token: str) -> bool: log.warning("The refresh token has expired, a new login is required.") return False self.access_token = json["access_token"] - self.expiry_time = datetime.datetime.utcnow() + datetime.timedelta( - seconds=json["expires_in"] - ) + self.expiry_time = datetime.datetime.now( + datetime.timezone.utc + ) + datetime.timedelta(seconds=json["expires_in"]) self.token_type = json["token_type"] # TIDAL rotates refresh tokens: the response may carry a new one and the # old one is then invalidated. Keep it, otherwise a persisted session ends From 8942dee53711448a3fac0f226d1a7512c5fb0cc3 Mon Sep 17 00:00:00 2001 From: Johannes Linde Date: Wed, 14 Oct 2026 00:02:30 +0200 Subject: [PATCH 2/3] Move project metadata from [tool.poetry] to PEP 621 [project] table --- poetry.lock | 4 ++-- pyproject.toml | 38 ++++++++++++++++++++++---------------- 2 files changed, 24 insertions(+), 18 deletions(-) diff --git a/poetry.lock b/poetry.lock index eb1991f..ea5f36e 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1875,5 +1875,5 @@ type = ["pytest-mypy (>=1.0.1) ; platform_python_implementation != \"PyPy\""] [metadata] lock-version = "2.1" -python-versions = "^3.10" -content-hash = "d63b3ce845ab1d0786a70506273292fff755696bffd39eb04a599570e4f5c63c" +python-versions = ">=3.10,<4.0" +content-hash = "522108f36b5380070d783b73b3bfe632953ff2ef434770e6bf01a0af45b7d830" diff --git a/pyproject.toml b/pyproject.toml index b8e7924..8932f9d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,18 +1,17 @@ -[tool.poetry] +[project] name = "tidalapi" version = "0.9.0" description = "Unofficial API for TIDAL music streaming service." -authors = ["Thomas Amland "] -maintainers = ["tehkillerbee "] +authors = [{ name = "Thomas Amland", email = "thomas.amland@googlemail.com" }] +maintainers = [ + { name = "tehkillerbee", email = "tehkillerbee@users.noreply.github.com" }, +] license = "LGPL-3.0-or-later" -readme = ["README.rst", "HISTORY.rst"] -homepage = "https://tidalapi.netlify.app" -repository = "https://github.com/EbbLabs/python-tidal" -documentation = "https://tidalapi.netlify.app" +license-files = ["LICENSE"] +requires-python = ">=3.10,<4.0" classifiers = [ "Development Status :: 4 - Beta", "Intended Audience :: Developers", - "License :: OSI Approved :: GNU Lesser General Public License v3 (LGPLv3)", "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", @@ -21,13 +20,21 @@ classifiers = [ "Programming Language :: Python :: 3.14", "Typing :: Typed", ] +dependencies = [ + "requests>=2.32.3,<3.0.0", + "python-dateutil>=2.8.2,<3.0.0", + "isodate>=0.7.2,<0.8.0", + "mpegdash>=0.4.1,<0.5.0", +] +dynamic = ["readme"] -[tool.poetry.dependencies] -python = "^3.10" -requests = "^2.32.3" -python-dateutil = "^2.8.2" -isodate = "^0.7.2" -mpegdash = "^0.4.1" +[project.urls] +Homepage = "https://tidalapi.netlify.app" +Repository = "https://github.com/EbbLabs/python-tidal" +Documentation = "https://tidalapi.netlify.app" + +[tool.poetry] +readme = ["README.rst", "HISTORY.rst"] [tool.poetry.group.dev.dependencies] mypy = "^2.4.0" @@ -45,7 +52,7 @@ docformatter = { extras = ["tomli"], version = "^1.7.3" } pytest-mock = "^3.11.1" [build-system] -requires = ["poetry-core"] +requires = ["poetry-core>=2.0.0,<3.0.0"] build-backend = "poetry.core.masonry.api" [tool.mypy] @@ -60,7 +67,6 @@ recursive = true black = true [tool.ruff] -target-version = "py310" line-length = 88 [tool.ruff.lint] From 36118388f0f43893cfc2446438a41887fda6ae84 Mon Sep 17 00:00:00 2001 From: Johannes Linde Date: Wed, 14 Oct 2026 00:05:58 +0200 Subject: [PATCH 3/3] Enable ruff pyupgrade (UP) rules and modernize syntax for python 3.10+ --- pyproject.toml | 2 +- tests/conftest.py | 11 ++- tests/test_album.py | 7 +- tests/test_user.py | 5 +- tidalapi/album.py | 88 ++++++++++++----------- tidalapi/artist.py | 73 +++++++++---------- tidalapi/genre.py | 14 ++-- tidalapi/media.py | 116 +++++++++++++++---------------- tidalapi/mix.py | 56 +++++++-------- tidalapi/page.py | 82 ++++++++++------------ tidalapi/playlist.py | 109 ++++++++++++++--------------- tidalapi/request.py | 42 +++++------ tidalapi/session.py | 110 ++++++++++++++--------------- tidalapi/types.py | 4 +- tidalapi/user.py | 162 +++++++++++++++++++++---------------------- tidalapi/workers.py | 2 +- 16 files changed, 427 insertions(+), 456 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 8932f9d..a5c62c8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -70,4 +70,4 @@ black = true line-length = 88 [tool.ruff.lint] -select = ["E4", "E7", "E9", "F"] +select = ["E4", "E7", "E9", "F", "UP"] diff --git a/tests/conftest.py b/tests/conftest.py index bbc22bb..41491dc 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -21,7 +21,6 @@ from json import dumps, loads from os import getenv from pathlib import Path -from typing import List, Optional import keyring import keyring.backends @@ -38,7 +37,7 @@ def session(request): class Credentials(ABC): - def load(self, key: str) -> Optional[dict]: + def load(self, key: str) -> dict | None: """Load secret.""" def save(self, key: str, val: dict) -> None: @@ -46,7 +45,7 @@ def save(self, key: str, val: dict) -> None: class EnvCredentials(Credentials): - def load(self, key: str) -> Optional[dict]: + def load(self, key: str) -> dict | None: credentials = { "token_type": getenv("TIDAL_TOKEN_TYPE", "Bearer"), "access_token": getenv("TIDAL_ACCESS_TOKEN"), @@ -64,7 +63,7 @@ def save(self, key: str, val: dict) -> None: class KeyringCredentials(Credentials): - def load(self, key: str) -> Optional[dict]: + def load(self, key: str) -> dict | None: with suppress(Exception): credentials = keyring.get_password(key, key) return loads(credentials) @@ -82,7 +81,7 @@ def __init__(self): self.cache_dir = base_dir / "python-tidal" self.cache_dir.mkdir(exist_ok=True) - def load(self, key: str) -> Optional[dict]: + def load(self, key: str) -> dict | None: cachef = self.cache_dir / f"{key}.json" try: return loads(cachef.read_text()) @@ -99,7 +98,7 @@ def save(self, key: str, val: dict) -> None: def get_credential_store( datastore_key: str = KEY, -) -> tuple[List[Credentials], Optional[dict]]: +) -> tuple[list[Credentials], dict | None]: stores = [] for store in (EnvCredentials, CachedCredentials, KeyringCredentials): with suppress(Exception): diff --git a/tests/test_album.py b/tests/test_album.py index 6f9dcc9..6d57cf2 100644 --- a/tests/test_album.py +++ b/tests/test_album.py @@ -140,10 +140,9 @@ def test_no_release_date(session): def test_default_image_not_used_on_albums_with_cover_art(session): album = session.album(108043414) assert album.cover is not None - default_album_url = "https://resources.tidal.com/images/%s/%ix%i.jpg" % ( - tidalapi.album.DEFAULT_ALBUM_IMG.replace("-", "/"), - 1280, - 1280, + default_album_path = tidalapi.album.DEFAULT_ALBUM_IMG.replace("-", "/") + default_album_url = ( + f"https://resources.tidal.com/images/{default_album_path}/1280x1280.jpg" ) # Album should not use default album art assert album.image(1280) != default_album_url diff --git a/tests/test_user.py b/tests/test_user.py index 3cfae37..9432a1d 100644 --- a/tests/test_user.py +++ b/tests/test_user.py @@ -796,10 +796,7 @@ def add_remove(object_id, add, remove, objects): model = type(item).__name__ name = item.name if exists: - reason = ( - "%s '%s' is already favorited, skipping to avoid changing the date it was favorited" - % (model, name) - ) + reason = f"{model} '{name}' is already favorited, skipping to avoid changing the date it was favorited" pytest.skip(reason) current_time = datetime.datetime.now(tz=dateutil.tz.tzutc()) diff --git a/tidalapi/album.py b/tidalapi/album.py index 29c581f..c1eb8d3 100644 --- a/tidalapi/album.py +++ b/tidalapi/album.py @@ -18,7 +18,7 @@ import copy import functools from datetime import datetime -from typing import TYPE_CHECKING, List, Optional, Union, cast +from typing import TYPE_CHECKING, Optional, Union, cast import dateutil.parser @@ -42,43 +42,43 @@ class Album: name, cover and video cover. TIDAL does this to reduce the network load. """ - id: Optional[int] = -1 - name: Optional[str] = None + id: int | None = -1 + name: str | None = None cover = None video_cover = None type = None - duration: Optional[int] = -1 - available: Optional[bool] = False - ad_supported_ready: Optional[bool] = False - dj_ready: Optional[bool] = False - allow_streaming: Optional[bool] = False - premium_streaming_only: Optional[bool] = False - num_tracks: Optional[int] = -1 - num_videos: Optional[int] = -1 - num_volumes: Optional[int] = -1 - tidal_release_date: Optional[datetime] = None - release_date: Optional[datetime] = None + duration: int | None = -1 + available: bool | None = False + ad_supported_ready: bool | None = False + dj_ready: bool | None = False + allow_streaming: bool | None = False + premium_streaming_only: bool | None = False + num_tracks: int | None = -1 + num_videos: int | None = -1 + num_volumes: int | None = -1 + tidal_release_date: datetime | None = None + release_date: datetime | None = None copyright = None upc = None version = None - explicit: Optional[bool] = True - universal_product_number: Optional[int] = -1 - popularity: Optional[int] = -1 - user_date_added: Optional[datetime] = None - audio_quality: Optional[str] = "" - audio_modes: Optional[List[str]] = [""] - media_metadata_tags: Optional[List[str]] = [""] + explicit: bool | None = True + universal_product_number: int | None = -1 + popularity: int | None = -1 + user_date_added: datetime | None = None + audio_quality: str | None = "" + audio_modes: list[str] | None = [""] + media_metadata_tags: list[str] | None = [""] artist: Optional["Artist"] = None - artists: Optional[List["Artist"]] = None + artists: list["Artist"] | None = None # Direct URL to https://listen.tidal.com/album/ listen_url: str = "" # Direct URL to https://tidal.com/browse/album/ share_url: str = "" - def __init__(self, session: "Session", album_id: Optional[str]): + def __init__(self, session: "Session", album_id: str | None): self.session = session self.request = session.request self.artist = session.artist() @@ -86,9 +86,9 @@ def __init__(self, session: "Session", album_id: Optional[str]): if self.id: try: - request = self.request.request("GET", "albums/%s" % self.id) + request = self.request.request("GET", f"albums/{self.id}") except ObjectNotFound as e: - e.args = ("Album with id %s not found" % self.id,) + e.args = (f"Album with id {self.id} not found",) raise e except TooManyRequests as e: e.args = ("Album unavailable",) @@ -100,7 +100,7 @@ def parse( self, json_obj: JsonObj, artist: Optional["Artist"] = None, - artists: Optional[List["Artist"]] = None, + artists: list["Artist"] | None = None, ) -> "Album": if artists is None: # The artists can be missing or empty, e.g. for removed albums @@ -165,7 +165,7 @@ def parse( return copy.copy(self) @property - def year(self) -> Optional[int]: + def year(self) -> int | None: """Get the year using :class:`available_release_date` :return: An :any:`python:int` containing the year the track was released @@ -173,7 +173,7 @@ def year(self) -> Optional[int]: return self.available_release_date.year if self.available_release_date else None @property - def available_release_date(self) -> Optional[datetime]: + def available_release_date(self) -> datetime | None: """Get the release date if it's available, otherwise get the day it was released on TIDAL. @@ -188,10 +188,10 @@ def available_release_date(self) -> Optional[datetime]: def tracks( self, - limit: Optional[int] = None, + limit: int | None = None, offset: int = 0, sparse_album: bool = False, - ) -> List["Track"]: + ) -> list["Track"]: """Returns the tracks in classes album. :param limit: The amount of items you want returned. @@ -211,14 +211,14 @@ def tracks( ) tracks = self.request.map_request( - "albums/%s/tracks" % self.id, params, parse=parse_track_callable + f"albums/{self.id}/tracks", params, parse=parse_track_callable ) assert isinstance(tracks, list) - return cast(List["Track"], tracks) + return cast(list["Track"], tracks) def items( self, limit: int = 100, offset: int = 0, sparse_album: bool = False - ) -> List[Union["Track", "Video"]]: + ) -> list[Union["Track", "Video"]]: """Gets the first 'limit' tracks and videos in the album from TIDAL. :param limit: The number of items you want to retrieve @@ -238,13 +238,13 @@ def items( ) items = self.request.map_request( - "albums/%s/items" % self.id, params=params, parse=parse_media_callable + f"albums/{self.id}/items", params=params, parse=parse_media_callable ) assert isinstance(items, list) - return cast(List[Union["Track", "Video"]], items) + return cast(list[Union["Track", "Video"]], items) def image( - self, dimensions: Union[int, str] = 320, default: str = DEFAULT_ALBUM_IMG + self, dimensions: int | str = 320, default: str = DEFAULT_ALBUM_IMG ) -> str: """A url to an album image cover. @@ -255,7 +255,7 @@ def image( Valid resolutions: 80x80, 160x160, 320x320, 640x640, 1280x1280, 3000x3000 (origin) """ if dimensions not in [80, 160, 320, 640, 1280, "origin"]: - raise ValueError("Invalid resolution {0} x {0}".format(dimensions)) + raise ValueError(f"Invalid resolution {dimensions} x {dimensions}") if dimensions == "origin": if not self.cover: @@ -280,7 +280,7 @@ def image( dimensions, ) - def video(self, dimensions: Union[int, str] = 320) -> str: + def video(self, dimensions: int | str = 320) -> str: """Creates a url to an mp4 video cover for the album. Valid resolutions: 80x80, 160x160, 320x320, 640x640, 1280x1280 @@ -293,7 +293,7 @@ def video(self, dimensions: Union[int, str] = 320) -> str: raise AttributeError("This album does not have a video cover.") if dimensions not in [80, 160, 320, 640, 1280, "origin"]: - raise ValueError("Invalid resolution {0} x {0}".format(dimensions)) + raise ValueError(f"Invalid resolution {dimensions} x {dimensions}") if dimensions == "origin": return self.session.config.video_url_origin % ( @@ -314,7 +314,7 @@ def page(self) -> "Page": """ return self.session.page.get("pages/album", params={"albumId": self.id}) - def similar(self) -> List["Album"]: + def similar(self) -> list["Album"]: """Retrieve albums similar to the current one. MetadataNotAvailable is raised, when no similar albums exist. @@ -322,7 +322,7 @@ def similar(self) -> List["Album"]: :return: A :any:`list` of similar albums """ try: - request = self.request.request("GET", "albums/%s/similar" % self.id) + request = self.request.request("GET", f"albums/{self.id}/similar") except ObjectNotFound: raise MetadataNotAvailable("No similar albums exist for this album") except TooManyRequests as e: @@ -333,7 +333,7 @@ def similar(self) -> List["Album"]: request.json(), parse=self.session.parse_album ) assert isinstance(albums, list) - return cast(List["Album"], albums) + return cast(list["Album"], albums) def review(self) -> str: """Retrieve the album review. @@ -341,9 +341,7 @@ def review(self) -> str: :return: A :class:`str` containing the album review, with wimp links :raises: :class:`requests.HTTPError` if there isn't a review yet """ - review = self.request.request("GET", "albums/%s/review" % self.id).json()[ - "text" - ] + review = self.request.request("GET", f"albums/{self.id}/review").json()["text"] assert isinstance(review, str) return review diff --git a/tidalapi/artist.py b/tidalapi/artist.py index 89701ce..38be0e2 100644 --- a/tidalapi/artist.py +++ b/tidalapi/artist.py @@ -17,9 +17,10 @@ """A module containing information and functions related to TIDAL artists.""" import copy +from collections.abc import Mapping from datetime import datetime from enum import Enum -from typing import TYPE_CHECKING, List, Mapping, NoReturn, Optional, Union, cast +from typing import TYPE_CHECKING, NoReturn, Optional, cast from warnings import warn import dateutil.parser @@ -39,20 +40,20 @@ class Artist: - id: Optional[int] = -1 - name: Optional[str] = None - roles: Optional[List["Role"]] = None + id: int | None = -1 + name: str | None = None + roles: list["Role"] | None = None role: Optional["Role"] = None - picture: Optional[str] = None - user_date_added: Optional[datetime] = None - bio: Optional[str] = None + picture: str | None = None + user_date_added: datetime | None = None + bio: str | None = None # Direct URL to https://listen.tidal.com/artist/ listen_url: str = "" # Direct URL to https://tidal.com/browse/artist/ share_url: str = "" - def __init__(self, session: "Session", artist_id: Optional[str]): + def __init__(self, session: "Session", artist_id: str | None): """Initialize the :class:`Artist` object, given a TIDAL artist ID :param session: The current TIDAL :class:`Session` :param str artist_id: TIDAL artist ID :raises: Raises :class:`exceptions.ObjectNotFound`""" @@ -62,9 +63,9 @@ def __init__(self, session: "Session", artist_id: Optional[str]): if self.id: try: - request = self.request.request("GET", "artists/%s" % self.id) + request = self.request.request("GET", f"artists/{self.id}") except ObjectNotFound as e: - e.args = ("Artist with id %s not found" % self.id,) + e.args = (f"Artist with id {self.id} not found",) raise e except TooManyRequests as e: e.args = ("Artist unavailable",) @@ -87,7 +88,7 @@ def parse_artist(self, json_obj: JsonObj) -> "Artist": self.roles = None self.role = None if json_obj.get("type") or json_obj.get("artistTypes"): - roles: List["Role"] = [] + roles: list[Role] = [] for role in json_obj.get("artistTypes", [json_obj.get("type")]): roles.append(Role(role)) @@ -109,7 +110,7 @@ def parse_artist(self, json_obj: JsonObj) -> "Artist": return copy.copy(self) - def parse_artists(self, json_obj: List[JsonObj]) -> List["Artist"]: + def parse_artists(self, json_obj: list[JsonObj]) -> list["Artist"]: """Parses a list of TIDAL artists, returns a list of :class:`Artist` objects Made for use within the python tidalapi module. @@ -120,16 +121,16 @@ def parse_artists(self, json_obj: List[JsonObj]) -> List["Artist"]: return list(map(self.parse_artist, json_obj)) def _get_albums( - self, params: Optional[Mapping[str, Union[int, str, None]]] = None - ) -> List["Album"]: + self, params: Mapping[str, int | str | None] | None = None + ) -> list["Album"]: return cast( - List["Album"], + list["Album"], self.request.map_request( f"artists/{self.id}/albums", params, parse=self.session.parse_album ), ) - def get_albums(self, limit: Optional[int] = None, offset: int = 0) -> List["Album"]: + def get_albums(self, limit: int | None = None, offset: int = 0) -> list["Album"]: """Queries TIDAL for the artists albums. :return: A list of :class:`Albums` @@ -138,8 +139,8 @@ def get_albums(self, limit: Optional[int] = None, offset: int = 0) -> List["Albu return self._get_albums(params) def get_albums_ep_singles( - self, limit: Optional[int] = None, offset: int = 0 - ) -> List["Album"]: + self, limit: int | None = None, offset: int = 0 + ) -> list["Album"]: warn( "This method is deprecated an will be removed in a future release. Use instead `get_ep_singles`", DeprecationWarning, @@ -149,8 +150,8 @@ def get_albums_ep_singles( return self.get_ep_singles(limit=limit, offset=offset) def get_ep_singles( - self, limit: Optional[int] = None, offset: int = 0 - ) -> List["Album"]: + self, limit: int | None = None, offset: int = 0 + ) -> list["Album"]: """Queries TIDAL for the artists extended plays and singles. :return: A list of :class:`Albums ` @@ -159,8 +160,8 @@ def get_ep_singles( return self._get_albums(params) def get_albums_other( - self, limit: Optional[int] = None, offset: int = 0 - ) -> List["Album"]: + self, limit: int | None = None, offset: int = 0 + ) -> list["Album"]: warn( "This method is deprecated an will be removed in a future release. Use instead `get_other`", DeprecationWarning, @@ -169,7 +170,7 @@ def get_albums_other( return self.get_other(limit=limit, offset=offset) - def get_other(self, limit: Optional[int] = None, offset: int = 0) -> List["Album"]: + def get_other(self, limit: int | None = None, offset: int = 0) -> list["Album"]: """Queries TIDAL for albums the artist has appeared on as a featured artist. :return: A list of :class:`Albums ` @@ -178,15 +179,15 @@ def get_other(self, limit: Optional[int] = None, offset: int = 0) -> List["Album return self._get_albums(params) def get_top_tracks( - self, limit: Optional[int] = None, offset: int = 0 - ) -> List["Track"]: + self, limit: int | None = None, offset: int = 0 + ) -> list["Track"]: """Queries TIDAL for the artists tracks, sorted by popularity. :return: A list of :class:`Tracks ` """ params = {"limit": limit, "offset": offset} return cast( - List["Track"], + list["Track"], self.request.map_request( f"artists/{self.id}/toptracks", params=params, @@ -194,14 +195,14 @@ def get_top_tracks( ), ) - def get_videos(self, limit: Optional[int] = None, offset: int = 0) -> List["Video"]: + def get_videos(self, limit: int | None = None, offset: int = 0) -> list["Video"]: """Queries tidal for the artists videos. :return: A list of :class:`Videos ` """ params = {"limit": limit, "offset": offset} return cast( - List["Video"], + list["Video"], self.request.map_request( f"artists/{self.id}/videos", params=params, @@ -219,19 +220,19 @@ def get_bio(self) -> str: str, self.request.request("GET", f"artists/{self.id}/bio").json()["text"] ) - def get_similar(self) -> List["Artist"]: + def get_similar(self) -> list["Artist"]: """Queries TIDAL for similar artists. :return: A list of :class:`Artists ` """ return cast( - List["Artist"], + list["Artist"], self.request.map_request( f"artists/{self.id}/similar", parse=self.session.parse_artist ), ) - def get_radio(self, limit: int = 100) -> List["Track"]: + def get_radio(self, limit: int = 100) -> list["Track"]: """Queries TIDAL for the artist radio, i.e. a list of tracks similar to this artist. @@ -241,7 +242,7 @@ def get_radio(self, limit: int = 100) -> List["Track"]: try: request = self.request.request( - "GET", "artists/%s/radio" % self.id, params=params + "GET", f"artists/{self.id}/radio", params=params ) except ObjectNotFound: raise MetadataNotAvailable("Track radio not available for this track") @@ -252,7 +253,7 @@ def get_radio(self, limit: int = 100) -> List["Track"]: json_obj = request.json() radio = self.request.map_json(json_obj, parse=self.session.parse_track) assert isinstance(radio, list) - return cast(List["Track"], radio) + return cast(list["Track"], radio) def get_radio_mix(self) -> mix.Mix: """Queries TIDAL for the artist radio, i.e. mix of tracks that are similar to @@ -263,7 +264,7 @@ def get_radio_mix(self) -> mix.Mix: available """ try: - request = self.request.request("GET", "artists/%s/mix" % self.id) + request = self.request.request("GET", f"artists/{self.id}/mix") except ObjectNotFound: raise MetadataNotAvailable("Artist radio not available for this artist") except TooManyRequests as e: @@ -273,7 +274,7 @@ def get_radio_mix(self) -> mix.Mix: json_obj = request.json() return self.session.mix(json_obj.get("id")) - def items(self) -> List[NoReturn]: + def items(self) -> list[NoReturn]: """The artist page does not supply any items. This only exists for symmetry with other model types. @@ -292,7 +293,7 @@ def image(self, dimensions: int = 320) -> str: Valid resolutions: 160x160, 320x320, 480x480, 750x750 """ if dimensions not in [160, 320, 480, 750]: - raise ValueError("Invalid resolution {0} x {0}".format(dimensions)) + raise ValueError(f"Invalid resolution {dimensions} x {dimensions}") if not self.picture: json = self.request.request("GET", f"artists/{self.id}").json() diff --git a/tidalapi/genre.py b/tidalapi/genre.py index 94ec0f1..e7e66ce 100644 --- a/tidalapi/genre.py +++ b/tidalapi/genre.py @@ -17,7 +17,7 @@ """""" import copy -from typing import TYPE_CHECKING, Any, List, Optional, cast +from typing import TYPE_CHECKING, Any, cast from tidalapi.types import JsonObj @@ -52,20 +52,20 @@ def parse_genre(self, json_obj: JsonObj) -> "Genre": return copy.copy(self) - def parse_genres(self, json_obj: List[JsonObj]) -> List["Genre"]: + def parse_genres(self, json_obj: list[JsonObj]) -> list["Genre"]: return list(map(self.parse_genre, json_obj)) - def get_genres(self) -> List["Genre"]: + def get_genres(self) -> list["Genre"]: return self.parse_genres(self.requests.request("GET", "genres").json()) - def items(self, model: List[Optional[Any]]) -> List[Optional[Any]]: + def items(self, model: list[Any | None]) -> list[Any | None]: """Gets the current genre's items of the specified type :param model: The tidalapi model you want returned. See :class:`Genre` :return: """ - type_relations: "TypeRelation" = next( + type_relations: TypeRelation = next( x for x in self.session.type_conversions if x.type == model ) name = type_relations.identifier @@ -73,6 +73,6 @@ def items(self, model: List[Optional[Any]]) -> List[Optional[Any]]: if getattr(self, name): location = f"genres/{self.path}/{name}" return cast( - List[Optional[Any]], self.requests.map_request(location, parse=parse) + list[Any | None], self.requests.map_request(location, parse=parse) ) - raise TypeError("This genre does not contain {0}".format(name)) + raise TypeError(f"This genre does not contain {name}") diff --git a/tidalapi/media.py b/tidalapi/media.py index cbe42b4..2ff35e9 100644 --- a/tidalapi/media.py +++ b/tidalapi/media.py @@ -25,7 +25,7 @@ from abc import abstractmethod from datetime import datetime, timedelta from enum import Enum -from typing import TYPE_CHECKING, Dict, List, Optional, Union, cast +from typing import TYPE_CHECKING, cast import dateutil.parser @@ -194,28 +194,26 @@ class Media: dj_ready: bool = False ad_supported_stream_ready: bool = False - stream_start_date: Optional[datetime] = None - tidal_release_date: Optional[datetime] = None # aka. streamStartDate - date_added: Optional[datetime] = None - user_date_added: Optional[datetime] = None # aka. dateAdded + stream_start_date: datetime | None = None + tidal_release_date: datetime | None = None # aka. streamStartDate + date_added: datetime | None = None + user_date_added: datetime | None = None # aka. dateAdded track_num: int = 1 # trackNumber volume_num: int = 1 # volumeNumber - artist: Optional["tidalapi.artist.Artist"] = None + artist: tidalapi.artist.Artist | None = None #: For the artist credit page artist_roles = None - artists: Optional[List["tidalapi.artist.Artist"]] = None - album: Optional["tidalapi.album.Album"] = None - type: Optional[str] = None + artists: list[tidalapi.artist.Artist] | None = None + album: tidalapi.album.Album | None = None + type: str | None = None # Direct URL to media https://listen.tidal.com/track/ or https://listen.tidal.com/browse/album//track/ listen_url: str = "" # Direct URL to media https://tidal.com/browse/track/ share_url: str = "" - def __init__( - self, session: "tidalapi.session.Session", media_id: Optional[str] = None - ): + def __init__(self, session: tidalapi.session.Session, media_id: str | None = None): self.session = session self.requests = self.session.request self.album = session.album() @@ -229,7 +227,7 @@ def _get(self, media_id: str) -> Media: "You are not supposed to use the media class directly." ) - def parse(self, json_obj: JsonObj, album: Optional[Album] = None) -> None: + def parse(self, json_obj: JsonObj, album: Album | None = None) -> None: """Assigns all :param json_obj: :param json_obj: The JSON object to parse @@ -288,8 +286,8 @@ def parse(self, json_obj: JsonObj, album: Optional[Album] = None) -> None: self.artist_roles = json_obj.get("artistRoles") def parse_media( - self, json_obj: JsonObj, album: Optional[Album] = None - ) -> Union["Track", "Video"]: + self, json_obj: JsonObj, album: Album | None = None + ) -> Track | Video: """Selects the media type when checking lists that can contain both. :param json_obj: The json containing the media @@ -314,19 +312,19 @@ class Track(Media): upload: bool = False # Audio quality and metadata - audio_quality: Optional[str] = None - audio_modes: Optional[List[str]] = None - media_metadata_tags: Optional[List[str]] = None + audio_quality: str | None = None + audio_modes: list[str] | None = None + media_metadata_tags: list[str] | None = None # Identification - index: Optional[int] = None - item_uuid: Optional[str] = None - isrc: Optional[str] = None + index: int | None = None + item_uuid: str | None = None + isrc: str | None = None # Track info - date_added: Optional[datetime] = None - description: Optional[str] = None - version: Optional[str] = None + date_added: datetime | None = None + description: str | None = None + version: str | None = None copyright: str = "" url: str = "" bpm: int = 0 @@ -335,12 +333,12 @@ class Track(Media): peak: float = 0.0 replay_gain: float = 0.0 # Related mixes - mixes: Optional[Dict] = None + mixes: dict | None = None # Derived fields: Full name from title, name and version full_name: str = "" - def parse_track(self, json_obj: JsonObj, album: Optional[Album] = None) -> Track: + def parse_track(self, json_obj: JsonObj, album: Album | None = None) -> Track: Media.parse(self, json_obj, album) self.pay_to_stream = json_obj.get("payToStream", False) self.premium_streaming_only = json_obj.get("premiumStreamingOnly", False) @@ -389,7 +387,7 @@ def parse_track(self, json_obj: JsonObj, album: Optional[Album] = None) -> Track return copy.copy(self) - def _get(self, media_id: str) -> "Track": + def _get(self, media_id: str) -> Track: """Returns information about a track, and also replaces the track used to call this function. @@ -397,9 +395,9 @@ def _get(self, media_id: str) -> "Track": :return: A :class:`Track` object containing all the information about the track """ try: - request = self.requests.request("GET", "tracks/%s" % media_id) + request = self.requests.request("GET", f"tracks/{media_id}") except ObjectNotFound as e: - e.args = ("Track with id %s not found" % media_id,) + e.args = (f"Track with id {media_id} not found",) raise e except TooManyRequests as e: e.args = ("Track unavailable",) @@ -419,9 +417,7 @@ def get_url(self) -> str: """ if self.session.is_pkce: raise URLNotAvailable( - "Track URL not available with quality:'{}'".format( - self.session.config.quality - ) + f"Track URL not available with quality:'{self.session.config.quality}'" ) params = { "urlusagemode": "STREAM", @@ -430,7 +426,7 @@ def get_url(self) -> str: } try: request = self.requests.request( - "GET", "tracks/%s/urlpostpaywall" % self.id, params + "GET", f"tracks/{self.id}/urlpostpaywall", params ) except ObjectNotFound: raise URLNotAvailable("URL not available for this track") @@ -441,14 +437,14 @@ def get_url(self) -> str: json_obj = request.json() return cast(str, json_obj["urls"][0]) - def lyrics(self) -> "Lyrics": + def lyrics(self) -> Lyrics: """Retrieves the lyrics for a song. :return: A :class:`Lyrics` object containing the lyrics :raises: A :class:`exceptions.MetadataNotAvailable` if there aren't any lyrics """ try: - request = self.requests.request("GET", "tracks/%s/lyrics" % self.id) + request = self.requests.request("GET", f"tracks/{self.id}/lyrics") except ObjectNotFound: raise MetadataNotAvailable("No lyrics exists for this track") except TooManyRequests as e: @@ -460,7 +456,7 @@ def lyrics(self) -> "Lyrics": assert not isinstance(lyrics, list) return cast("Lyrics", lyrics) - def get_track_radio(self, limit: int = 100) -> List["Track"]: + def get_track_radio(self, limit: int = 100) -> list[Track]: """Queries TIDAL for the track radio mix as a list of tracks similar to this track. @@ -472,7 +468,7 @@ def get_track_radio(self, limit: int = 100) -> List["Track"]: try: request = self.requests.request( - "GET", "tracks/%s/radio" % self.id, params=params + "GET", f"tracks/{self.id}/radio", params=params ) except ObjectNotFound: raise MetadataNotAvailable("Track radio not available for this track") @@ -483,7 +479,7 @@ def get_track_radio(self, limit: int = 100) -> List["Track"]: json_obj = request.json() tracks = self.requests.map_json(json_obj, parse=self.session.parse_track) assert isinstance(tracks, list) - return cast(List["Track"], tracks) + return cast(list["Track"], tracks) def get_radio_mix(self) -> mix.Mix: """Queries TIDAL for the track radio mix of tracks that are similar to this @@ -494,7 +490,7 @@ def get_radio_mix(self) -> mix.Mix: available """ try: - request = self.requests.request("GET", "tracks/%s/mix" % self.id) + request = self.requests.request("GET", f"tracks/{self.id}/mix") except ObjectNotFound: raise MetadataNotAvailable("Track radio not available for this track") except TooManyRequests as e: @@ -504,7 +500,7 @@ def get_radio_mix(self) -> mix.Mix: json_obj = request.json() return self.session.mix(json_obj.get("id")) - def get_stream(self) -> "Stream": + def get_stream(self) -> Stream: """Retrieves the track streaming object, allowing for audio transmission. :return: A :class:`Stream` object which holds audio file properties and @@ -520,7 +516,7 @@ def get_stream(self) -> "Stream": try: request = self.requests.request( - "GET", "tracks/%s/playbackinfopostpaywall" % self.id, params + "GET", f"tracks/{self.id}/playbackinfopostpaywall", params ) except ObjectNotFound: raise StreamNotAvailable("Stream not available for this track") @@ -575,7 +571,7 @@ class Stream: bit_depth: int = 16 sample_rate: int = 44100 - def parse(self, json_obj: JsonObj) -> "Stream": + def parse(self, json_obj: JsonObj) -> Stream: self.track_id = json_obj.get("trackId") self.audio_mode = json_obj.get("audioMode") self.audio_quality = json_obj.get("audioQuality") @@ -598,7 +594,7 @@ def parse(self, json_obj: JsonObj) -> "Stream": def get_audio_resolution(self): return self.bit_depth, self.sample_rate - def get_stream_manifest(self) -> "StreamManifest": + def get_stream_manifest(self) -> StreamManifest: return StreamManifest(self) def get_manifest_data(self) -> str: @@ -690,7 +686,7 @@ def get_sampling_rate(self) -> int: return self.dash_info.audio_sampling_rate @staticmethod - def get_mimetype(stream_codec, stream_url: Optional[str] = None) -> str: + def get_mimetype(stream_codec, stream_url: str | None = None) -> str: if stream_codec: return MimeType.from_audio_codec(stream_codec) if not stream_url: @@ -702,7 +698,7 @@ def get_mimetype(stream_codec, stream_url: Optional[str] = None) -> str: return MimeType.audio_m4a @staticmethod - def get_file_extension(stream_url: str, stream_codec: Optional[str] = None) -> str: + def get_file_extension(stream_url: str, stream_codec: str | None = None) -> str: if AudioExtensions.FLAC in stream_url: # If the file extension within the URL is '*.flac', this is simply a FLAC file. result: str = AudioExtensions.FLAC @@ -753,7 +749,7 @@ class DashInfo: urls: [str] = [""] @staticmethod - def from_stream(stream) -> "DashInfo": + def from_stream(stream) -> DashInfo: try: if stream.is_mpd and not stream.is_encrypted: return DashInfo(stream.get_manifest_data()) @@ -761,7 +757,7 @@ def from_stream(stream) -> "DashInfo": raise ManifestDecodeError from e @staticmethod - def from_mpd(mpd_manifest) -> "DashInfo": + def from_mpd(mpd_manifest) -> DashInfo: try: return DashInfo(mpd_manifest) except Exception as e: @@ -856,7 +852,7 @@ def get_urls(mpd) -> list[str]: def get_hls(self) -> str: hls = "#EXTM3U\n" - hls += "#EXT-X-TARGETDURATION:%s\n" % int(self.duration.seconds) + hls += f"#EXT-X-TARGETDURATION:{int(self.duration.seconds)}\n" hls += "#EXT-X-VERSION:3\n" items = self.urls chunk_duration = "#EXTINF:%0.3f,\n" % ( @@ -883,7 +879,7 @@ class Lyrics: subtitles: str = "" right_to_left: bool = False - def parse(self, json_obj: JsonObj) -> "Lyrics": + def parse(self, json_obj: JsonObj) -> Lyrics: self.track_id = json_obj["trackId"] self.provider = json_obj["lyricsProvider"] self.provider_track_id = json_obj["providerCommontrackId"] @@ -898,17 +894,17 @@ def parse(self, json_obj: JsonObj) -> "Lyrics": class Video(Media): """An object containing information about a video.""" - release_date: Optional[datetime] = None - video_quality: Optional[str] = None - image_id: Optional[str] = None - image_path: Optional[str] = None - cover: Optional[str] = None # aka. image_id + release_date: datetime | None = None + video_quality: str | None = None + image_id: str | None = None + image_path: str | None = None + cover: str | None = None # aka. image_id vibrant_color: str = "#000000" ads_pre_paywall_only: bool = False - ads_url: Optional[str] = "" + ads_url: str | None = "" - def parse_video(self, json_obj: JsonObj, album: Optional[Album] = None) -> Video: + def parse_video(self, json_obj: JsonObj, album: Album | None = None) -> Video: Media.parse(self, json_obj, album) release_date = json_obj.get("releaseDate") self.release_date = ( @@ -942,9 +938,9 @@ def _get(self, media_id: str) -> Video: unavailable """ try: - request = self.requests.request("GET", "videos/%s" % self.id) + request = self.requests.request("GET", f"videos/{self.id}") except ObjectNotFound as e: - e.args = ("Video with id %s not found" % media_id,) + e.args = (f"Video with id {media_id} not found",) raise e except TooManyRequests as e: e.args = ("Video unavailable",) @@ -970,7 +966,7 @@ def get_url(self) -> str: try: request = self.requests.request( - "GET", "videos/%s/urlpostpaywall" % self.id, params + "GET", f"videos/{self.id}/urlpostpaywall", params ) except ObjectNotFound: raise URLNotAvailable("URL not available for this video") @@ -983,7 +979,7 @@ def get_url(self) -> str: def image(self, width: int = 1080, height: int = 720) -> str: if (width, height) not in [(160, 107), (480, 320), (750, 500), (1080, 720)]: - raise ValueError("Invalid resolution {} x {}".format(width, height)) + raise ValueError(f"Invalid resolution {width} x {height}") if not self.cover: raise AttributeError("No cover image") return self.session.config.image_url % ( diff --git a/tidalapi/mix.py b/tidalapi/mix.py index 819d19b..b330235 100644 --- a/tidalapi/mix.py +++ b/tidalapi/mix.py @@ -21,7 +21,7 @@ from dataclasses import dataclass from datetime import datetime from enum import Enum -from typing import TYPE_CHECKING, List, Optional, Union +from typing import TYPE_CHECKING import dateutil.parser @@ -68,20 +68,20 @@ class Mix: title: str = "" sub_title: str = "" sharing_images = None - mix_type: Optional[MixType] = None + mix_type: MixType | None = None content_behaviour: str = "" short_subtitle: str = "" - images: Optional[ImageResponse] = None + images: ImageResponse | None = None _retrieved = False - _items: Optional[List[Union["Video", "Track"]]] = None + _items: list[Video | Track] | None = None - def __init__(self, session: Session, mix_id: Optional[str]): + def __init__(self, session: Session, mix_id: str | None): self.session = session self.request = session.request if mix_id is not None: self.get(mix_id) - def get(self, mix_id: Optional[str] = None) -> "Mix": + def get(self, mix_id: str | None = None) -> Mix: """Returns information about a mix, and also replaces the mix object used to call this function. @@ -96,7 +96,7 @@ def get(self, mix_id: Optional[str] = None) -> "Mix": try: request = self.request.request("GET", "pages/mix", params=params) except ObjectNotFound as e: - e.args = ("Mix with id %s not found" % mix_id,) + e.args = (f"Mix with id {mix_id} not found",) raise e except TooManyRequests as e: e.args = ("Mix unavailable",) @@ -113,7 +113,7 @@ def get(self, mix_id: Optional[str] = None) -> "Mix": self._items = result.categories[1].items return self - def parse(self, json_obj: JsonObj) -> "Mix": + def parse(self, json_obj: JsonObj) -> Mix: """Parse a mix into a :class:`Mix`, replaces the calling object. :param json_obj: The json of a mix to be parsed @@ -135,7 +135,7 @@ def parse(self, json_obj: JsonObj) -> "Mix": return copy.copy(self) - def items(self) -> List[Union["Video", "Track"]]: + def items(self) -> list[Video | Track]: """Returns all the items in the mix, retrieves them with :class:`get` as well if not already done. @@ -178,25 +178,25 @@ class TextInfo: class MixV2: """A mix from TIDALs v2 api endpoint.""" - mix_type: Optional[MixType] = None - country_code: Optional[str] = None - date_added: Optional[datetime] = None - id: Optional[str] = None - artifact_id_type: Optional[str] = None - content_behavior: Optional[str] = None - images: Optional[ImageResponse] = None - detail_images: Optional[ImageResponse] = None + mix_type: MixType | None = None + country_code: str | None = None + date_added: datetime | None = None + id: str | None = None + artifact_id_type: str | None = None + content_behavior: str | None = None + images: ImageResponse | None = None + detail_images: ImageResponse | None = None master = False is_stable_id = False - title: Optional[str] = None - sub_title: Optional[str] = None - short_subtitle: Optional[str] = None - title_text_info: Optional[TextInfo] = None - sub_title_text_info: Optional[TextInfo] = None - short_subtitle_text_info: Optional[TextInfo] = None - updated: Optional[datetime] = None + title: str | None = None + sub_title: str | None = None + short_subtitle: str | None = None + title_text_info: TextInfo | None = None + sub_title_text_info: TextInfo | None = None + short_subtitle_text_info: TextInfo | None = None + updated: datetime | None = None _retrieved = False - _items: Optional[List[Union["Video", "Track"]]] = None + _items: list[Video | Track] | None = None def __init__(self, session: Session, mix_id: str): self.session = session @@ -204,7 +204,7 @@ def __init__(self, session: Session, mix_id: str): if mix_id is not None: self.get(mix_id) - def get(self, mix_id: Optional[str] = None) -> "MixV2": + def get(self, mix_id: str | None = None) -> MixV2: """Returns information about a mix, and also replaces the mix object used to call this function. @@ -218,7 +218,7 @@ def get(self, mix_id: Optional[str] = None) -> "MixV2": try: request = self.request.request("GET", "pages/mix", params=params) except ObjectNotFound as e: - e.args = ("Mix with id %s not found" % mix_id,) + e.args = (f"Mix with id {mix_id} not found",) raise e except TooManyRequests as e: e.args = ("Mix unavailable",) @@ -236,7 +236,7 @@ def get(self, mix_id: Optional[str] = None) -> "MixV2": self._items = result.categories[1].items return self - def parse(self, json_obj: JsonObj) -> "MixV2": + def parse(self, json_obj: JsonObj) -> MixV2: """Parse a mix into a :class:`MixV2`, replaces the calling object. :param json_obj: The json of a mix to be parsed diff --git a/tidalapi/page.py b/tidalapi/page.py index 8647ec2..1fbbff9 100644 --- a/tidalapi/page.py +++ b/tidalapi/page.py @@ -19,16 +19,12 @@ import copy import logging +from collections.abc import Callable, Iterator from dataclasses import dataclass from typing import ( TYPE_CHECKING, Any, - Callable, - Dict, - Iterator, - List, Optional, - Type, Union, cast, ) @@ -64,7 +60,7 @@ "HorizontalListWithContext", ] -AllCategoriesV2 = Union[PageCategoriesV2] +AllCategoriesV2 = PageCategoriesV2 log = logging.getLogger(__name__) @@ -78,11 +74,9 @@ class Page: """ title: str = "" - categories: Optional[List[Union["AllCategories", "AllCategoriesV2"]]] = None - _categories_iter: Optional[Iterator[Union["AllCategories", "AllCategoriesV2"]]] = ( - None - ) - _items_iter: Optional[Iterator[Callable[..., Any]]] = None + categories: list[Union["AllCategories", "AllCategoriesV2"]] | None = None + _categories_iter: Iterator[Union["AllCategories", "AllCategoriesV2"]] | None = None + _items_iter: Iterator[Callable[..., Any]] | None = None page_category: "PageCategory" page_category_v2: "PageCategoryV2" request: "Requests" @@ -99,7 +93,7 @@ def __iter__(self) -> "Page": raise AttributeError("No categories found") self._categories_iter = iter(self.categories) self._category = next(self._categories_iter) - self._items_iter = iter(cast(List[Callable[..., Any]], self._category.items)) + self._items_iter = iter(cast(list[Callable[..., Any]], self._category.items)) return self def __next__(self) -> Callable[..., Any]: @@ -112,7 +106,7 @@ def __next__(self) -> Callable[..., Any]: raise AttributeError("No categories found") self._category = next(self._categories_iter) self._items_iter = iter( - cast(List[Callable[..., Any]], self._category.items) + cast(list[Callable[..., Any]], self._category.items) ) return self.__next__() return item @@ -138,7 +132,7 @@ def parse(self, json_obj: JsonObj) -> "Page": return copy.copy(self) - def get(self, endpoint: str, params: Optional[Dict[str, Any]] = None) -> "Page": + def get(self, endpoint: str, params: dict[str, Any] | None = None) -> "Page": """Retrieve a page from the specified endpoint, overwrites the calling page. :param params: Parameter to retrieve the page with @@ -175,15 +169,15 @@ def parse(cls, json_obj: JsonObj) -> Optional["More"]: class PageCategory: type = None - title: Optional[str] = None - description: Optional[str] = "" + title: str | None = None + description: str | None = "" request: "Requests" - _more: Optional[More] = None + _more: More | None = None def __init__(self, session: "Session"): self.session = session self.request = session.request - self.item_types: Dict[str, Callable[..., Any]] = { + 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, @@ -228,7 +222,7 @@ def parse(self, json_obj: JsonObj) -> AllCategories: return category.parse(json_obj) - def show_more(self) -> Optional[Page]: + def show_more(self) -> Page | None: """Get the full list of items on their own :class:`.Page` from a :class:`.PageCategory` @@ -251,14 +245,14 @@ class PageCategoryV2: """ # Registry mapping 'type' strings to subclass types - _type_map: Dict[str, Type["PageCategoryV2"]] = {} + _type_map: dict[str, type["PageCategoryV2"]] = {} # Common metadata fields for all category types - type: Optional[str] = None - module_id: Optional[str] = None - title: Optional[str] = None - subtitle: Optional[str] = None - description: Optional[str] = "" + type: str | None = None + module_id: str | None = None + title: str | None = None + subtitle: str | None = None + description: str | None = "" _more: Optional["More"] = None def __init__(self, session: "Session"): @@ -271,7 +265,7 @@ def __init__(self, session: "Session"): self.request = session.request # Common item parsers by type (can be used by subclasses like SimpleList) - self.item_types: Dict[str, Callable[..., Any]] = { + self.item_types: dict[str, Callable[..., Any]] = { "PLAYLIST": self.session.parse_playlist, "VIDEO": self.session.parse_video, "TRACK": self.session.parse_track, @@ -297,7 +291,7 @@ def decorator(subclass): return decorator - def parse_item(self, list_item: Dict) -> "PageCategoryV2": + def parse_item(self, list_item: dict) -> "PageCategoryV2": """Factory method that creates the correct subclass instance based on the 'type' field in item Dict, parses base fields, and then calls subclass parse().""" category_type = list_item.get("type") @@ -309,7 +303,7 @@ def parse_item(self, list_item: Dict) -> "PageCategoryV2": instance.parse(list_item) return instance - def _parse_base(self, list_item: Dict): + def _parse_base(self, list_item: dict): """Parse fields common to all categories.""" self.type = list_item.get("type") self.module_id = list_item.get("moduleId") @@ -322,7 +316,7 @@ def parse(self, json_obj: JsonObj): """Subclasses implement this method to parse category-specific data.""" raise NotImplementedError("Subclasses must implement parse()") - def view_all(self) -> Optional[Page]: + def view_all(self) -> Page | None: """View all items in a Get the full list of items on their own :class:`.Page` from a :class:`.PageCategory` @@ -339,7 +333,7 @@ class SimpleList(PageCategoryV2): def __init__(self, session: "Session"): super().__init__(session) - self.items: List[Any] = [] + self.items: list[Any] = [] def parse(self, json_obj: "JsonObj"): self.items = [ @@ -378,7 +372,7 @@ class TrackList(PageCategoryV2): def __init__(self, session: "Session"): super().__init__(session) - self.items: List[Any] = [] + self.items: list[Any] = [] def parse(self, json_obj: "JsonObj"): self.items = [ @@ -391,7 +385,7 @@ def parse(self, json_obj: "JsonObj"): class FeaturedItems(PageCategory): """Items that have been featured by TIDAL.""" - items: Optional[List["PageItem"]] = None + items: list["PageItem"] | None = None def __init__(self, session: "Session"): super().__init__(session) @@ -410,7 +404,7 @@ def parse(self, json_obj: JsonObj) -> "FeaturedItems": class PageLinks(PageCategory): """A list of :class:`.PageLink` to other parts of TIDAL.""" - items: Optional[List["PageLink"]] = None + items: list["PageLink"] | None = None def parse(self, json_obj: JsonObj) -> "PageLinks": """Parse the list of links from TIDAL. @@ -431,7 +425,7 @@ 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: Optional[List[Any]] = None + items: list[Any] | None = None def parse(self, json_obj: JsonObj) -> "ItemList": """Parse a list of items on TIDAL from the pages endpoints. @@ -443,8 +437,8 @@ def parse(self, json_obj: JsonObj) -> "ItemList": self.title = json_obj["title"] item_type = json_obj["type"] list_key = "pagedList" - session: Optional["Session"] = None - parse: Optional[Callable[..., Any]] = None + session: Session | None = None + parse: Callable[..., Any] | None = None if item_type in self.item_types.keys(): parse = self.item_types[item_type] @@ -459,7 +453,7 @@ def parse(self, json_obj: JsonObj) -> "ItemList": item["item"]["artistRoles"] = item["roles"] session = self.session else: - raise NotImplementedError("PageType {} not implemented".format(item_type)) + raise NotImplementedError(f"PageType {item_type} not implemented") self.items = self.request.map_json(json_obj[list_key], parse, session) @@ -541,13 +535,13 @@ def get( raise NotImplementedError(f"PageItem type {self.type} not implemented") -class TextBlock(object): +class TextBlock: """A block of text, with a named icon, which seems to be left up to the application.""" text: str = "" icon: str = "" - items: Optional[List[str]] = None + items: list[str] | None = None def __init__(self, session: "Session"): self.session = session @@ -563,9 +557,9 @@ def parse(self, json_obj: JsonObj) -> "TextBlock": class LinkList(PageCategory): """A list of items containing links, e.g. social links or articles.""" - items: Optional[List[Any]] = None - title: Optional[str] = None - description: Optional[str] = None + items: list[Any] | None = None + title: str | None = None + description: str | None = None def parse(self, json_obj: JsonObj) -> "LinkList": self.items = json_obj["items"] @@ -575,10 +569,10 @@ def parse(self, json_obj: JsonObj) -> "LinkList": return copy.copy(self) -class ItemHeader(object): +class ItemHeader: """Single item in a "category" of the page.""" - items: Optional[List[Any]] = None + items: list[Any] | None = None def __init__(self, item: Any): self.items = [item] diff --git a/tidalapi/playlist.py b/tidalapi/playlist.py index c421a37..c9620be 100644 --- a/tidalapi/playlist.py +++ b/tidalapi/playlist.py @@ -19,8 +19,9 @@ from __future__ import annotations import copy +from collections.abc import Sequence from datetime import datetime -from typing import TYPE_CHECKING, List, Optional, Sequence, Union, cast +from typing import TYPE_CHECKING, Union, cast from tidalapi.exceptions import ObjectNotFound, TooManyRequests from tidalapi.types import ItemOrder, JsonObj, OrderDirection @@ -48,32 +49,32 @@ class Playlist: """An object containing various data about a playlist and methods to work with them.""" - id: Optional[str] = None - trn: Optional[str] = None - name: Optional[str] = None + id: str | None = None + trn: str | None = None + name: str | None = None num_tracks: int = -1 num_videos: int = -1 - creator: Optional[Union["Artist", "User"]] = None - description: Optional[str] = None + creator: Artist | User | None = None + description: str | None = None duration: int = -1 - last_updated: Optional[datetime] = None - created: Optional[datetime] = None + last_updated: datetime | None = None + created: datetime | None = None type = None - public: Optional[bool] = False - popularity: Optional[int] = None - promoted_artists: Optional[List["Artist"]] = None - last_item_added_at: Optional[datetime] = None - picture: Optional[str] = None - square_picture: Optional[str] = None - user_date_added: Optional[datetime] = None - _etag: Optional[str] = None + public: bool | None = False + popularity: int | None = None + promoted_artists: list[Artist] | None = None + last_item_added_at: datetime | None = None + picture: str | None = None + square_picture: str | None = None + user_date_added: datetime | None = None + _etag: str | None = None # Direct URL to https://listen.tidal.com/playlist/ listen_url: str = "" # Direct URL to https://tidal.com/browse/playlist/ share_url: str = "" - def __init__(self, session: "Session", playlist_id: Optional[str]): + def __init__(self, session: Session, playlist_id: str | None): self.id = playlist_id self.session = session self.request = session.request @@ -82,7 +83,7 @@ def __init__(self, session: "Session", playlist_id: Optional[str]): try: request = self.request.request("GET", self._base_url % self.id) except ObjectNotFound as e: - e.args = ("Playlist with id %s not found" % playlist_id,) + e.args = (f"Playlist with id {playlist_id} not found",) raise e except TooManyRequests as e: e.args = ("Playlist unavailable",) @@ -91,7 +92,7 @@ def __init__(self, session: "Session", playlist_id: Optional[str]): self._etag = request.headers["etag"] self.parse(request.json()) - def parse(self, obj: JsonObj) -> "Playlist": + def parse(self, obj: JsonObj) -> Playlist: """Parses a playlist from tidal, replaces the current playlist object. :param obj: Json data returned from api.tidal.com containing a playlist @@ -149,7 +150,7 @@ def parse(self, obj: JsonObj) -> "Playlist": return copy.copy(self) - def factory(self) -> Union["Playlist", "UserPlaylist"]: + def factory(self) -> Playlist | UserPlaylist: if ( self.id and self.creator @@ -160,7 +161,7 @@ def factory(self) -> Union["Playlist", "UserPlaylist"]: return self - def parse_factory(self, json_obj: JsonObj) -> "Playlist": + def parse_factory(self, json_obj: JsonObj) -> Playlist: self.parse(json_obj) return copy.copy(self.factory()) @@ -200,11 +201,11 @@ def get_items_count( def tracks( self, - limit: Optional[int] = None, + limit: int | None = None, offset: int = 0, - order: Optional[ItemOrder] = None, - order_direction: Optional[OrderDirection] = None, - ) -> List["Track"]: + order: ItemOrder | None = None, + order_direction: OrderDirection | None = None, + ) -> list[Track]: """Gets the playlists' tracks from TIDAL. :param limit: The amount of items you want returned. @@ -236,9 +237,9 @@ def tracks( def tracks_paginated( self, - order: Optional[ItemOrder] = None, - order_direction: Optional[OrderDirection] = None, - ) -> List["Playlist"]: + order: ItemOrder | None = None, + order_direction: OrderDirection | None = None, + ) -> list[Playlist]: """Get the tracks in the playlist, using pagination. :param order: Optional; A :class:`ItemOrder` describing the ordering type when @@ -255,9 +256,9 @@ def items( self, limit: int = 100, offset: int = 0, - order: Optional[ItemOrder] = None, - order_direction: Optional[OrderDirection] = None, - ) -> List[Union["Track", "Video"]]: + order: ItemOrder | None = None, + order_direction: OrderDirection | None = None, + ) -> list[Track | Video]: """Fetches up to the first 100 items, including tracks and videos. :param limit: The amount of items you want, up to 100. @@ -294,7 +295,7 @@ def image(self, dimensions: int = 480, wide_fallback: bool = True) -> str: Original sizes: 160x160, 320x320, 480x480, 640x640, 750x750, 1080x1080 """ if dimensions not in [160, 320, 480, 640, 750, 1080]: - raise ValueError("Invalid resolution {0} x {0}".format(dimensions)) + raise ValueError(f"Invalid resolution {dimensions} x {dimensions}") if self.square_picture: return self.session.config.image_url % ( self.square_picture.replace("-", "/"), @@ -316,7 +317,7 @@ def wide_image(self, width: int = 1080, height: int = 720) -> str: Valid sizes: 160x107, 480x320, 750x500, 1080x720 """ if (width, height) not in [(160, 107), (480, 320), (750, 500), (1080, 720)]: - raise ValueError("Invalid resolution {} x {}".format(width, height)) + raise ValueError(f"Invalid resolution {width} x {height}") if self.picture is None: raise AttributeError("No picture available") return self.session.config.image_url % ( @@ -330,14 +331,14 @@ class Folder: """An object containing various data about a folder and methods to work with them.""" - trn: Optional[str] = None - id: Optional[str] = None - parent_folder_id: Optional[str] = None - name: Optional[str] = None - parent: Optional[str] = None # TODO Determine the correct type of the parent - added: Optional[datetime] = None - created: Optional[datetime] = None - last_modified: Optional[datetime] = None + trn: str | None = None + id: str | None = None + parent_folder_id: str | None = None + name: str | None = None + parent: str | None = None # TODO Determine the correct type of the parent + added: datetime | None = None + created: datetime | None = None + last_modified: datetime | None = None total_number_of_items: int = 0 # Direct URL to https://listen.tidal.com/folder/ @@ -345,8 +346,8 @@ class Folder: def __init__( self, - session: "Session", - folder_id: Optional[str], + session: Session, + folder_id: str | None, parent_folder_id: str = "root", ): self.id = folder_id @@ -401,7 +402,7 @@ def _reparse(self) -> None: self.parse(item) return - def parse(self, json_obj: JsonObj) -> "Folder": + def parse(self, json_obj: JsonObj) -> Folder: """Parses a folder from tidal, replaces the current folder object. :param json_obj: Json data returned from api.tidal.com containing a folder @@ -454,9 +455,7 @@ def remove(self) -> bool: params=params, ).ok - def items( - self, offset: int = 0, limit: int = 50 - ) -> List[Union["Playlist", "UserPlaylist"]]: + def items(self, offset: int = 0, limit: int = 50) -> list[Playlist | UserPlaylist]: """Return the items in the folder. :param offset: Optional; The index of the first item to be returned. Default: 0 @@ -481,7 +480,7 @@ def items( if json_obj.get("items"): playlists = {"items": [item["data"] for item in json_obj.get("items")]} return cast( - List[Union["Playlist", "UserPlaylist"]], + list[Union["Playlist", "UserPlaylist"]], self.request.map_json(playlists, parse=self.playlist.parse_factory), ) else: @@ -539,9 +538,7 @@ def _reparse(self) -> None: self._etag = request.headers["etag"] self.request.map_json(request.json(), parse=self.parse) - def edit( - self, title: Optional[str] = None, description: Optional[str] = None - ) -> bool: + def edit(self, title: str | None = None, description: str | None = None) -> bool: """Edit UserPlaylist title & description. :param title: Playlist title @@ -556,7 +553,7 @@ def edit( data = {"title": title, "description": description} return self.request.request("POST", self._base_url % self.id, data=data).ok - def delete_by_id(self, media_ids: List[str]) -> bool: + def delete_by_id(self, media_ids: list[str]) -> bool: """Delete one or more items from the UserPlaylist. :param media_ids: Lists of Media IDs to remove. @@ -570,11 +567,11 @@ def delete_by_id(self, media_ids: List[str]) -> bool: def add( self, - media_ids: List[str], + media_ids: list[str], allow_duplicates: bool = False, position: int = -1, limit: int = 100, - ) -> List[int]: + ) -> list[int]: """Add one or more items to the UserPlaylist. :param media_ids: List of Media IDs to add. @@ -613,7 +610,7 @@ def add( def merge( self, playlist: str, allow_duplicates: bool = False, allow_missing: bool = True - ) -> List[int]: + ) -> list[int]: """Add (merge) items from a playlist with the current playlist. :param playlist: Playlist UUID to be merged in the current playlist @@ -815,4 +812,4 @@ def delete(self) -> bool: :return: True, if successful. """ - return self.request.request("DELETE", path="playlists/%s" % self.id).ok + return self.request.request("DELETE", path=f"playlists/{self.id}").ok diff --git a/tidalapi/request.py b/tidalapi/request.py index a935880..78af56e 100644 --- a/tidalapi/request.py +++ b/tidalapi/request.py @@ -21,16 +21,12 @@ import logging import threading import time +from collections.abc import Callable, Mapping, MutableMapping from typing import ( TYPE_CHECKING, Any, - Callable, - List, Literal, - Mapping, - MutableMapping, Optional, - Union, cast, ) from urllib.parse import urljoin @@ -48,7 +44,7 @@ log = logging.getLogger(__name__) -Params = Mapping[str, Union[str, int, None]] +Params = Mapping[str, str | int | None] Methods = Literal["GET", "POST", "PUT", "DELETE"] @@ -65,7 +61,7 @@ TOKEN_EXPIRY_MARGIN = datetime.timedelta(seconds=60) -class Requests(object): +class Requests: """A class for handling api requests to TIDAL.""" user_agent: str @@ -86,7 +82,7 @@ def __init__(self, session: "Session"): self._backoff_until = 0.0 self._backoff_reason = "" - def _refresh_access_token(self, stale_access_token: Optional[str]) -> bool: + def _refresh_access_token(self, stale_access_token: str | None) -> bool: """Refresh the access token, at most once for concurrent callers. Requests are often issued from several threads (e.g. the paginated workers). If @@ -169,8 +165,8 @@ def _send( self, method: Methods, url: str, - request_params: Mapping[str, Union[str, int, None]], - data: Optional[JsonObj], + request_params: Mapping[str, str | int | None], + data: JsonObj | None, headers: MutableMapping[str, str], ) -> requests.Response: """Send a request with the session's current access token.""" @@ -188,10 +184,10 @@ def basic_request( self, method: Methods, path: str, - params: Optional[Params] = None, - data: Optional[JsonObj] = None, - headers: Optional[MutableMapping[str, str]] = None, - base_url: Optional[str] = None, + params: Params | None = None, + data: JsonObj | None = None, + headers: MutableMapping[str, str] | None = None, + base_url: str | None = None, ) -> requests.Response: self._raise_if_backing_off() request_params = { @@ -258,10 +254,10 @@ def request( self, method: Methods, path: str, - params: Optional[Params] = None, - data: Optional[JsonObj] = None, - headers: Optional[MutableMapping[str, str]] = None, - base_url: Optional[str] = None, + params: Params | None = None, + data: JsonObj | None = None, + headers: MutableMapping[str, str] | None = None, + base_url: str | None = None, ) -> requests.Response: """Method for tidal requests. @@ -280,7 +276,7 @@ def request( try: request.raise_for_status() except requests.HTTPError as e: - log.info("Request resulted in exception {}".format(e)) + log.info(f"Request resulted in exception {e}") self.latest_err_response = request if err := http_error_to_tidal_error(e): raise err from e @@ -314,8 +310,8 @@ def get_latest_err_response_str(self) -> str: def map_request( self, url: str, - params: Optional[Params] = None, - parse: Optional[Callable[..., Any]] = None, + params: Params | None = None, + parse: Callable[..., Any] | None = None, ) -> Any: """Returns the data about object(s) at the specified url, with the method specified in the parse argument. @@ -339,7 +335,7 @@ def map_request( def map_json( cls, json_obj: JsonObj, - parse: Optional[Callable[..., Any]] = None, + parse: Callable[..., Any] | None = None, session: Optional["Session"] = None, ) -> Any: items = json_obj.get("items") @@ -356,7 +352,7 @@ def map_json( for item in items: item["item"]["dateAdded"] = item["created"] - lists: List[Any] = [] + lists: list[Any] = [] for item in items: if session is not None: parse = cast( diff --git a/tidalapi/session.py b/tidalapi/session.py index a854fd3..2843a3b 100644 --- a/tidalapi/session.py +++ b/tidalapi/session.py @@ -27,19 +27,15 @@ import random import time import uuid +from collections.abc import Callable from dataclasses import dataclass from enum import Enum from pathlib import Path from typing import ( TYPE_CHECKING, Any, - Callable, - List, Literal, - Optional, - Tuple, TypedDict, - Union, cast, no_type_check, ) @@ -62,7 +58,7 @@ from tidalapi.user import FetchedUser, LoggedInUser, PlaylistCreator log = logging.getLogger(__name__) -SearchTypes: List[Optional[Any]] = [ +SearchTypes: list[Any | None] = [ artist.Artist, album.Album, media.Track, @@ -199,9 +195,9 @@ class Case(Enum): scream = id lower = id - identifier: List[str] - type: List[Union[object, None]] - parse: List[Callable[..., Any]] + identifier: list[str] + type: list[object | None] + parse: list[Callable[..., Any]] TypeConversionKeys = Literal["identifier", "type", "parse"] @@ -210,38 +206,38 @@ class Case(Enum): @dataclass class TypeRelation: identifier: str - type: Optional[Any] + type: Any | None parse: Callable[..., Any] 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]] + artists: list[artist.Artist] + albums: list[album.Album] + tracks: list[media.Track] + videos: list[media.Video] + playlists: list[playlist.Playlist | playlist.UserPlaylist] + top_hit: list[Any] | None class Session: """Object for interacting with the TIDAL api and.""" #: The TIDAL access token, this is what you use with load_oauth_session - access_token: Optional[str] = None + access_token: str | None = None #: A :class:`datetime` object containing the date the access token will expire - expiry_time: Optional[datetime.datetime] = None + expiry_time: datetime.datetime | None = None #: A refresh token for retrieving a new access token through refresh_token - refresh_token: Optional[str] = None + refresh_token: str | None = None #: The type of access token, e.g. Bearer - token_type: Optional[str] = None + token_type: str | None = None #: The session id for a TIDAL session, you also need this to use load_oauth_session - session_id: Optional[str] = None - country_code: Optional[str] = None - locale: Optional[str] = None + session_id: str | None = None + country_code: str | None = None + locale: str | None = None #: A :class:`.User` object containing the currently logged in user. - user: Optional[Union["FetchedUser", "LoggedInUser", "PlaylistCreator"]] = None + user: FetchedUser | LoggedInUser | PlaylistCreator | None = None - def __init__(self, config: Optional[Config] = None): + def __init__(self, config: Config | None = None): self.config = config if config is not None else Config() self.request_session = requests.Session() @@ -255,9 +251,9 @@ def __init__(self, config: Optional[Config] = None): self.is_pkce = False # True if current session is PKCE type, otherwise false # The (client_id, client_secret) in use before `client_enable_hires` replaced them - self._device_client: Optional[Tuple[str, str]] = None + self._device_client: tuple[str, str] | None = None - self.type_conversions: List[TypeRelation] = [ + self.type_conversions: list[TypeRelation] = [ TypeRelation( identifier=identifier, type=type, parse=cast(Callable[..., Any], parse) ) @@ -287,7 +283,7 @@ def parse_album(self, obj: JsonObj) -> album.Album: return self.album().parse(obj) def parse_track( - self, obj: JsonObj, album: Optional[album.Album] = None + self, obj: JsonObj, album: album.Album | None = None ) -> media.Track: """Parse an album from the given response.""" return self.track().parse_track(obj, album) @@ -297,8 +293,8 @@ def parse_video(self, obj: JsonObj) -> media.Video: return self.video().parse_video(obj) def parse_media( - self, obj: JsonObj, album: Optional[album.Album] = None - ) -> Union[media.Track, media.Video]: + self, obj: JsonObj, album: album.Album | None = None + ) -> media.Track | media.Video: """Parse a media type (track, video) from the given response.""" return self.track().parse_media(obj, album) @@ -306,7 +302,7 @@ def parse_artist(self, obj: JsonObj) -> artist.Artist: """Parse an artist from the given response.""" return self.artist().parse_artist(obj) - def parse_artists(self, obj: List[JsonObj]) -> List[artist.Artist]: + def parse_artists(self, obj: list[JsonObj]) -> list[artist.Artist]: """Parse an artist from the given response.""" return self.artist().parse_artists(obj) @@ -334,7 +330,7 @@ def convert_type( output: TypeConversionKeys = "identifier", case: Case = Case.lower, suffix: bool = True, - ) -> Union[str, Callable[..., Any]]: + ) -> str | Callable[..., Any]: type_relations = next( x for x in self.type_conversions if getattr(x, search_type) == search ) @@ -354,8 +350,8 @@ def convert_type( def load_session( self, session_id: str, - country_code: Optional[str] = None, - user_id: Optional[int] = None, + country_code: str | None = None, + user_id: int | None = None, ) -> bool: """Establishes TIDAL login details using a previous session id. @@ -387,9 +383,9 @@ def load_oauth_session( self, token_type: str, access_token: str, - refresh_token: Optional[str] = None, - expiry_time: Optional[datetime.datetime] = None, - is_pkce: Optional[bool] = False, + refresh_token: str | None = None, + expiry_time: datetime.datetime | None = None, + is_pkce: bool | None = False, ) -> bool: """Login to TIDAL using details from a previous OAuth login, automatically refreshes expired access tokens if refresh_token is supplied as well. @@ -423,7 +419,7 @@ def load_oauth_session( def login_session_file( self, session_file: Path, - do_pkce: Optional[bool] = False, + do_pkce: bool | None = False, fn_print: Callable[[str], None] = print, ) -> bool: """Logs in to the TIDAL api using an existing OAuth/PKCE session file. @@ -485,7 +481,7 @@ def login_pkce(self, fn_print: Callable[[str], None] = print) -> None: # Get redirect URL from user input. url_redirect: str = input("Paste 'Ooops' page URL here and press :") # Query for auth tokens - json: dict[str, Union[str, int]] = self.pkce_get_auth_token(url_redirect) + json: dict[str, str | int] = self.pkce_get_auth_token(url_redirect) # Parse and set tokens. self.process_auth_token(json, is_pkce_token=True) @@ -505,7 +501,7 @@ def client_enable_hires(self): self.config.client_id = self.config.client_id_pkce self.config.client_secret = self.config.client_secret_pkce - def _device_client_credentials(self) -> Tuple[str, str]: + def _device_client_credentials(self) -> tuple[str, str]: """The client_id and client_secret for the limited input device login.""" if self._device_client: return self._device_client @@ -531,7 +527,7 @@ def pkce_login_url(self) -> str: return self.config.api_pkce_auth + "?" + urlencode(params) - def pkce_get_auth_token(self, url_redirect: str) -> dict[str, Union[str, int]]: + def pkce_get_auth_token(self, url_redirect: str) -> dict[str, str | int]: """Parses the redirect url to extract access and refresh tokens. :param url_redirect: URL of the 'Ooops' page, where the user was redirected to @@ -569,7 +565,7 @@ def pkce_get_auth_token(self, url_redirect: str) -> dict[str, Union[str, int]]: # Parse the JSON response. try: - token: dict[str, Union[str, int]] = response.json() + token: dict[str, str | int] = response.json() except requests.exceptions.JSONDecodeError: raise Exception("Wrong one-time authorization code", response) @@ -588,7 +584,7 @@ def login_oauth_simple(self, fn_print: Callable[[str], None] = print) -> None: fn_print(text.format(login.verification_uri_complete, login.expires_in)) future.result() - def login_oauth(self) -> Tuple[LinkLogin, concurrent.futures.Future[Any]]: + def login_oauth(self) -> tuple[LinkLogin, concurrent.futures.Future[Any]]: """Login to TIDAL with a remote link for limited input devices. The function will return everything you need to log in through a web browser, and will return @@ -690,7 +686,7 @@ def process_link_login( return result_process def process_auth_token( - self, json: dict[str, Union[str, int]], is_pkce_token: bool = True + self, json: dict[str, str | int], is_pkce_token: bool = True ) -> bool: """Parses the authorization response and sets the token values to the specific variables for further usage. @@ -827,7 +823,7 @@ def video_quality(self, quality: str) -> None: def search( self, query: str, - models: Optional[List[Optional[Any]]] = None, + models: list[Any | None] | None = None, limit: int = 50, offset: int = 0, ) -> SearchResults: @@ -849,7 +845,7 @@ def search( if not models: models = SearchTypes - types: List[str] = [] + types: list[str] = [] # This converts the specified TIDAL models in the models list into the text versions so we can parse it. for model in models: if model not in SearchTypes: @@ -891,12 +887,12 @@ def check_login(self) -> bool: if self.user is None or not self.user.id or not self.session_id: return False return self.request.basic_request( - "GET", "users/%s/subscription" % self.user.id + "GET", f"users/{self.user.id}/subscription" ).ok def playlist( - self, playlist_id: Optional[str] = None - ) -> Union[playlist.Playlist, playlist.UserPlaylist]: + self, playlist_id: str | None = None + ) -> 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. @@ -912,7 +908,7 @@ def playlist( log.warning("Playlist '%s' is unavailable", playlist_id) raise - def folder(self, folder_id: Optional[str] = None) -> playlist.Folder: + def folder(self, folder_id: str | None = None) -> playlist.Folder: """Function to create a Folder object with access to the session instance in a smoother way. Calls :class:`tidalapi.Folder(session=session, folder_id=track_id) <.Folder>` internally. @@ -928,7 +924,7 @@ def folder(self, folder_id: Optional[str] = None) -> playlist.Folder: raise def track( - self, track_id: Optional[str] = None, with_album: bool = False + self, track_id: str | None = 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) @@ -1028,7 +1024,7 @@ def get_albums_by_barcode(self, barcode: str) -> list[album.Album]: log.error("API Response: '%s'", resp_str) raise InvalidUPC - def video(self, video_id: Optional[str] = None) -> media.Video: + def video(self, video_id: str | None = 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. @@ -1044,7 +1040,7 @@ def video(self, video_id: Optional[str] = None) -> media.Video: log.warning("Video '%s' is unavailable", video_id) raise - def artist(self, artist_id: Optional[str] = None) -> artist.Artist: + def artist(self, artist_id: str | None = 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. @@ -1060,7 +1056,7 @@ def artist(self, artist_id: Optional[str] = None) -> artist.Artist: log.warning("Artist '%s' is unavailable", artist_id) raise - def album(self, album_id: Optional[str] = None) -> album.Album: + def album(self, album_id: str | None = 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. @@ -1076,7 +1072,7 @@ def album(self, album_id: Optional[str] = None) -> album.Album: log.warning("Album '%s' is unavailable", album_id) raise - def mix(self, mix_id: Optional[str] = None) -> mix.Mix: + def mix(self, mix_id: str | None = 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. @@ -1108,8 +1104,8 @@ def mixv2(self, mix_id=None) -> mix.MixV2: raise def get_user( - self, user_id: Optional[int] = None - ) -> Union["FetchedUser", "LoggedInUser", "PlaylistCreator"]: + self, user_id: int | None = None + ) -> FetchedUser | LoggedInUser | 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. diff --git a/tidalapi/types.py b/tidalapi/types.py index 50b4711..426f756 100644 --- a/tidalapi/types.py +++ b/tidalapi/types.py @@ -1,9 +1,9 @@ # Copyright (C) 2023- The Tidalapi Developers from enum import Enum -from typing import Any, Dict +from typing import Any -JsonObj = Dict[str, Any] +JsonObj = dict[str, Any] class AlbumOrder(Enum): diff --git a/tidalapi/user.py b/tidalapi/user.py index e58f028..a616667 100644 --- a/tidalapi/user.py +++ b/tidalapi/user.py @@ -23,7 +23,7 @@ from __future__ import annotations from copy import copy -from typing import TYPE_CHECKING, List, Optional, Union, cast +from typing import TYPE_CHECKING, Union, cast from urllib.parse import urljoin from tidalapi.exceptions import ObjectNotFound @@ -69,26 +69,24 @@ class User: and id. """ - id: Optional[int] = -1 + id: int | None = -1 - def __init__(self, session: "Session", user_id: Optional[int]): + def __init__(self, session: Session, user_id: int | None): self.id = user_id self.session = session self.request = session.request self.playlist = session.playlist() self.folder = session.folder() - def factory(self) -> Union["LoggedInUser", "FetchedUser", "PlaylistCreator"]: + def factory(self) -> LoggedInUser | FetchedUser | PlaylistCreator: return cast( Union["LoggedInUser", "FetchedUser", "PlaylistCreator"], - self.request.map_request("users/%s" % self.id, parse=self.parse), + self.request.map_request(f"users/{self.id}", parse=self.parse), ) - def parse( - self, json_obj: JsonObj - ) -> Union["LoggedInUser", "FetchedUser", "PlaylistCreator"]: + def parse(self, json_obj: JsonObj) -> LoggedInUser | FetchedUser | PlaylistCreator: if "username" in json_obj: - user: Union[LoggedInUser, FetchedUser, PlaylistCreator] = LoggedInUser( + user: LoggedInUser | FetchedUser | PlaylistCreator = LoggedInUser( self.session, json_obj["id"] ) @@ -106,11 +104,11 @@ def parse( class FetchedUser(User): - first_name: Optional[str] = None - last_name: Optional[str] = None - picture_id: Optional[str] = None + first_name: str | None = None + last_name: str | None = None + picture_id: str | None = None - def parse(self, json_obj: JsonObj) -> "FetchedUser": + def parse(self, json_obj: JsonObj) -> FetchedUser: self.id = json_obj["id"] self.first_name = json_obj["firstName"] self.last_name = json_obj["lastName"] @@ -120,7 +118,7 @@ def parse(self, json_obj: JsonObj) -> "FetchedUser": def image(self, dimensions: int) -> str: if dimensions not in [100, 210, 600]: - raise ValueError("Invalid resolution {0} x {0}".format(dimensions)) + raise ValueError(f"Invalid resolution {dimensions} x {dimensions}") if self.picture_id is None: raise AttributeError("No picture available") @@ -133,39 +131,39 @@ def image(self, dimensions: int) -> str: class LoggedInUser(FetchedUser): - username: Optional[str] = None - email: Optional[str] = None - profile_metadata: Optional[JsonObj] = None + username: str | None = None + email: str | None = None + profile_metadata: JsonObj | None = None - def __init__(self, session: "Session", user_id: Optional[int]): - super(LoggedInUser, self).__init__(session, user_id) + def __init__(self, session: Session, user_id: int | None): + super().__init__(session, user_id) assert self.id is not None, "User is not logged in" self.favorites = Favorites(session, self.id) - def parse(self, json_obj: JsonObj) -> "LoggedInUser": - super(LoggedInUser, self).parse(json_obj) + def parse(self, json_obj: JsonObj) -> LoggedInUser: + super().parse(json_obj) self.username = json_obj["username"] self.email = json_obj["email"] self.profile_metadata = json_obj return copy(self) - def playlists(self) -> List[Union["Playlist", "UserPlaylist"]]: + def playlists(self) -> list[Playlist | UserPlaylist]: """Get the (personal) playlists created by the user. :return: Returns a list of :class:`~tidalapi.playlist.Playlist` objects containing the playlists. """ return cast( - List[Union["Playlist", "UserPlaylist"]], + list[Union["Playlist", "UserPlaylist"]], self.request.map_request( - "users/%s/playlists" % self.id, parse=self.playlist.parse_factory + f"users/{self.id}/playlists", parse=self.playlist.parse_factory ), ) def public_playlists( self, offset: int = 0, limit: int = 50 - ) -> List[Union["Playlist", "UserPlaylist"]]: + ) -> list[Playlist | UserPlaylist]: """Get the (public) playlists created by the user. :param limit: The number of items you want returned. @@ -173,7 +171,7 @@ def public_playlists( :return: List of public playlists. """ params = {"limit": limit, "offset": offset} - endpoint = "user-playlists/%s/public" % self.id + endpoint = f"user-playlists/{self.id}/public" json_obj = self.request.request( "GET", endpoint, base_url=self.session.config.api_v2_location, params=params ).json() @@ -185,13 +183,13 @@ def public_playlists( playlists["items"].append(item["playlist"]) return cast( - List[Union["Playlist", "UserPlaylist"]], + list[Union["Playlist", "UserPlaylist"]], self.request.map_json(playlists, parse=self.playlist.parse_factory), ) def playlist_and_favorite_playlists( self, offset: int = 0, limit: int = 50 - ) -> List[Union["Playlist", "UserPlaylist"]]: + ) -> list[Playlist | UserPlaylist]: """Get the playlists created by the user, and the playlists favorited by the user. This function is limited to 50 by TIDAL, requiring pagination. @@ -199,7 +197,7 @@ def playlist_and_favorite_playlists( containing the playlists. """ params = {"limit": limit, "offset": offset} - endpoint = "users/%s/playlistsAndFavoritePlaylists" % self.id + endpoint = f"users/{self.id}/playlistsAndFavoritePlaylists" json_obj = self.request.request("GET", endpoint, params=params).json() # This endpoint sorts them into favorited and created playlists, but we already do that when parsing them. @@ -208,13 +206,13 @@ def playlist_and_favorite_playlists( json_obj["items"][index] = item["playlist"] return cast( - List[Union["Playlist", "UserPlaylist"]], + list[Union["Playlist", "UserPlaylist"]], self.request.map_json(json_obj, parse=self.playlist.parse_factory), ) def create_playlist( self, title: str, description: str, parent_id: str = "root" - ) -> "UserPlaylist": + ) -> UserPlaylist: """Create a playlist in the specified parent folder. :param title: Playlist title @@ -239,7 +237,7 @@ def create_playlist( else: raise ObjectNotFound("Playlist not found after creation") - def create_folder(self, title: str, parent_id: str = "root") -> "Folder": + def create_folder(self, title: str, parent_id: str = "root") -> Folder: """Create folder in the specified parent folder. :param title: Folder title @@ -263,9 +261,9 @@ def create_folder(self, title: str, parent_id: str = "root") -> "Folder": class PlaylistCreator(User): - name: Optional[str] = None + name: str | None = None - def parse(self, json_obj: JsonObj) -> "PlaylistCreator": + def parse(self, json_obj: JsonObj) -> PlaylistCreator: if self.id == 0 or self.session.user is None: self.name = "TIDAL" @@ -284,7 +282,7 @@ def parse(self, json_obj: JsonObj) -> "PlaylistCreator": class Favorites: """An object containing a users favourites.""" - def __init__(self, session: "Session", user_id: int): + def __init__(self, session: Session, user_id: int): self.session = session self.requests = session.request self.base_url = f"users/{user_id}/favorites" @@ -557,9 +555,9 @@ def remove_folders_playlists( def artists_paginated( self, - order: Optional[ArtistOrder] = None, - order_direction: Optional[OrderDirection] = None, - ) -> List["Artist"]: + order: ArtistOrder | None = None, + order_direction: OrderDirection | None = None, + ) -> list[Artist]: """Get the users favorite artists, using pagination. :param order: Optional; A :class:`ArtistOrder` describing the ordering type when @@ -578,9 +576,9 @@ def artists( self, limit: int = 50, offset: int = 0, - order: Optional[ArtistOrder] = None, - order_direction: Optional[OrderDirection] = None, - ) -> List["Artist"]: + order: ArtistOrder | None = None, + order_direction: OrderDirection | None = None, + ) -> list[Artist]: """Get the users favorite artists. :param limit: The number of artist you want returned. @@ -599,7 +597,7 @@ def artists( params["orderDirection"] = order_direction.value return cast( - List["Artist"], + list["Artist"], self.requests.map_request( f"{self.base_url}/artists", params=params, @@ -624,9 +622,9 @@ def get_artists_count( def albums_paginated( self, - order: Optional[AlbumOrder] = None, - order_direction: Optional[OrderDirection] = None, - ) -> List["Album"]: + order: AlbumOrder | None = None, + order_direction: OrderDirection | None = None, + ) -> list[Album]: """Get the users favorite albums, using pagination. :param order: Optional; A :class:`AlbumOrder` describing the ordering type when @@ -645,9 +643,9 @@ def albums( self, limit: int = 50, offset: int = 0, - order: Optional[AlbumOrder] = None, - order_direction: Optional[OrderDirection] = None, - ) -> List["Album"]: + order: AlbumOrder | None = None, + order_direction: OrderDirection | None = None, + ) -> list[Album]: """Get the users favorite albums. :param limit: The number of albums you want returned. @@ -666,7 +664,7 @@ def albums( params["orderDirection"] = order_direction.value return cast( - List["Album"], + list["Album"], self.requests.map_request( f"{self.base_url}/albums", params=params, parse=self.session.parse_album ), @@ -689,9 +687,9 @@ def get_albums_count( def playlists_paginated( self, - order: Optional[PlaylistOrder] = None, - order_direction: Optional[OrderDirection] = None, - ) -> List["Playlist"]: + order: PlaylistOrder | None = None, + order_direction: OrderDirection | None = None, + ) -> list[Playlist]: """Get the users favorite playlists, using cursor-based pagination. The v2 my-collection/playlists/folders endpoint uses cursor-based pagination. @@ -705,8 +703,8 @@ def playlists_paginated( :return: A :class:`list` :class:`~tidalapi.playlist.Playlist` objects containing the favorite playlists. """ - playlists: List["Playlist"] = [] - cursor: Optional[str] = None + playlists: list[Playlist] = [] + cursor: str | None = None while True: items = self.playlists( @@ -725,10 +723,10 @@ def playlists( self, limit: int = 50, offset: int = 0, - order: Optional[PlaylistOrder] = None, - order_direction: Optional[OrderDirection] = None, - cursor: Optional[str] = None, - ) -> List["Playlist"]: + order: PlaylistOrder | None = None, + order_direction: OrderDirection | None = None, + cursor: str | None = None, + ) -> list[Playlist]: """Get the users favorite playlists (v2 endpoint), relative to the root folder. This function is limited to 50 by TIDAL, requiring pagination. @@ -768,17 +766,17 @@ def playlists( json_obj, parse=self.session.parse_playlist ) self._last_playlists_cursor = json_obj.get("cursor") - return cast(List["Playlist"], items) + return cast(list["Playlist"], items) def playlist_folders( self, limit: int = 50, offset: int = 0, - order: Optional[PlaylistOrder] = None, - order_direction: Optional[OrderDirection] = None, + order: PlaylistOrder | None = None, + order_direction: OrderDirection | None = None, parent_folder_id: str = "root", - cursor: Optional[str] = None, - ) -> List["Folder"]: + cursor: str | None = None, + ) -> list[Folder]: """Get a list of folders created by the user. :param limit: The number of playlists you want returned (Note: Cannot exceed 50) @@ -814,7 +812,7 @@ def playlist_folders( json_obj = self.session.request.request("GET", url, params).json() items = self.session.request.map_json(json_obj, parse=self.session.parse_folder) self._last_folders_cursor = json_obj.get("cursor") - return cast(List["Folder"], items) + return cast(list["Folder"], items) def get_playlists_count(self) -> int: """Get the total number of playlists in the user's root collection. @@ -838,9 +836,9 @@ def get_playlists_count(self) -> int: def tracks_paginated( self, - order: Optional[ItemOrder] = None, - order_direction: Optional[OrderDirection] = None, - ) -> List["Playlist"]: + order: ItemOrder | None = None, + order_direction: OrderDirection | None = None, + ) -> list[Playlist]: """Get the users favorite tracks, using pagination. :param order: Optional; A :class:`ItemOrder` describing the ordering type when @@ -859,9 +857,9 @@ def tracks( self, limit: int = 50, offset: int = 0, - order: Optional[ItemOrder] = None, - order_direction: Optional[OrderDirection] = None, - ) -> List["Track"]: + order: ItemOrder | None = None, + order_direction: OrderDirection | None = None, + ) -> list[Track]: """Get the users favorite tracks. :param limit: The number of tracks you want returned. @@ -880,7 +878,7 @@ def tracks( params["orderDirection"] = order_direction.value return cast( - List["Track"], + list["Track"], self.requests.map_request( f"{self.base_url}/tracks", params=params, parse=self.session.parse_track ), @@ -904,9 +902,9 @@ def get_tracks_count( def videos_paginated( self, - order: Optional[ItemOrder] = None, - order_direction: Optional[OrderDirection] = None, - ) -> List["Video"]: + order: ItemOrder | None = None, + order_direction: OrderDirection | None = None, + ) -> list[Video]: """Get the users favorite videos, using pagination. :param order: Optional; A :class:`ItemOrder` describing the ordering type when @@ -925,9 +923,9 @@ def videos( self, limit: int = 50, offset: int = 0, - order: Optional[VideoOrder] = None, - order_direction: Optional[OrderDirection] = None, - ) -> List["Video"]: + order: VideoOrder | None = None, + order_direction: OrderDirection | None = None, + ) -> list[Video]: """Get the users favorite videos. :param limit: The number of videos you want returned. @@ -946,7 +944,7 @@ def videos( params["orderDirection"] = order_direction.value return cast( - List["Video"], + list["Video"], self.requests.map_request( f"{self.base_url}/videos", params=params, @@ -973,9 +971,9 @@ def mixes( self, limit: int = 50, offset: int = 0, - order: Optional[MixOrder] = None, - order_direction: Optional[OrderDirection] = None, - ) -> List["MixV2"]: + order: MixOrder | None = None, + order_direction: OrderDirection | None = None, + ) -> list[MixV2]: """Get the users favorite mixes & radio. :param limit: The number of mixes you want returned. @@ -994,7 +992,7 @@ def mixes( params["orderDirection"] = order_direction.value return cast( - List["MixV2"], + list["MixV2"], self.requests.map_request( url=urljoin( self.session.config.api_v2_location, f"{self.v2_base_url}/mixes" diff --git a/tidalapi/workers.py b/tidalapi/workers.py index 48a3456..12ce040 100644 --- a/tidalapi/workers.py +++ b/tidalapi/workers.py @@ -1,6 +1,6 @@ import logging +from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor -from typing import Callable log = logging.getLogger(__name__)