Skip to content

Commit

Permalink
Improve, cleanup error handling. Cleanup misc imports. Added misc. fu…
Browse files Browse the repository at this point in the history
…nction descriptions
  • Loading branch information
tehkillerbee committed Feb 13, 2024
1 parent a5d86c5 commit e907654
Show file tree
Hide file tree
Showing 5 changed files with 125 additions and 62 deletions.
36 changes: 19 additions & 17 deletions tidalapi/album.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@

import dateutil.parser

from tidalapi.exceptions import MetadataNotAvailable, ObjectNotFound
from tidalapi.types import JsonObj

if TYPE_CHECKING:
Expand Down Expand Up @@ -68,17 +69,16 @@ class Album:

def __init__(self, session: "Session", album_id: Optional[str]):
self.session = session
self.requests = session.request
self.request = session.request
self.artist = session.artist()
self.id = album_id

if self.id:
json_obj = self.requests.map_request("albums/%s" % self.id)
if json_obj.get("status"):
assert json_obj.get("status") == 404
raise AttributeError("Album not found")
request = self.request.request("GET", "albums/%s" % self.id)
if request.status_code and request.status_code == 404:
raise ObjectNotFound("Album not found")
else:
self.requests.map_json(json_obj, parse=self.parse)
self.request.map_json(request.json(), parse=self.parse)

def parse(
self,
Expand Down Expand Up @@ -133,7 +133,7 @@ def parse(

@property
def year(self) -> Optional[int]:
"""Convenience function to get the year using :class:`available_release_date`
"""Get the year using :class:`available_release_date`
:return: An :any:`python:int` containing the year the track was released
"""
Expand Down Expand Up @@ -161,7 +161,8 @@ def tracks(self, limit: Optional[int] = None, offset: int = 0) -> List["Track"]:
:return: A list of the :class:`Tracks <.Track>` in the album.
"""
params = {"limit": limit, "offset": offset}
tracks = self.requests.map_request(

tracks = self.request.map_request(
"albums/%s/tracks" % self.id, params, parse=self.session.parse_track
)
assert isinstance(tracks, list)
Expand All @@ -175,7 +176,7 @@ def items(self, limit: int = 100, offset: int = 0) -> List[Union["Track", "Video
:return: A list of :class:`Tracks<.Track>` and :class:`Videos`<.Video>`
"""
params = {"offset": offset, "limit": limit}
items = self.requests.map_request(
items = self.request.map_request(
"albums/%s/items" % self.id, params=params, parse=self.session.parse_media
)
assert isinstance(items, list)
Expand Down Expand Up @@ -232,17 +233,18 @@ def page(self) -> "Page":
return self.session.page.get("pages/album", params={"albumId": self.id})

def similar(self) -> List["Album"]:
"""Retrieve albums similar to the current one. AttributeError is raised, when no
similar albums exists.
"""Retrieve albums similar to the current one. MetadataNotAvailable is raised,
when no similar albums exist.
:return: A :any:`list` of similar albums
"""
json_obj = self.requests.map_request("albums/%s/similar" % self.id)
if json_obj.get("status"):
assert json_obj.get("status") == 404
raise AttributeError("No similar albums exist for this album")
request = self.request.request("GET", "albums/%s/similar" % self.id)
if request.status_code and request.status_code == 404:
raise MetadataNotAvailable("No similar albums exist for this album")
else:
albums = self.requests.map_json(json_obj, parse=self.session.parse_album)
albums = self.request.map_json(
request.json(), parse=self.session.parse_album
)
assert isinstance(albums, list)
return cast(List["Album"], albums)

Expand All @@ -253,7 +255,7 @@ def review(self) -> str:
:raises: :class:`requests.HTTPError` if there isn't a review yet
"""
# morguldir: TODO: Add parsing of wimplinks?
review = self.requests.request("GET", "albums/%s/review" % self.id).json()[
review = self.request.request("GET", "albums/%s/review" % self.id).json()[
"text"
]
assert isinstance(review, str)
Expand Down
26 changes: 18 additions & 8 deletions tidalapi/artist.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import dateutil.parser
from typing_extensions import NoReturn

from tidalapi.exceptions import ObjectNotFound
from tidalapi.types import JsonObj

if TYPE_CHECKING:
Expand All @@ -45,17 +46,26 @@ class Artist:
bio: Optional[str] = None

def __init__(self, session: "Session", artist_id: Optional[str]):
"""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`"""
self.session = session
self.request = self.session.request
self.id = artist_id

if self.id:
self.request.map_request(f"artists/{artist_id}", parse=self.parse_artist)
request = self.request.request("GET", "artists/%s" % self.id)
if request.status_code and request.status_code == 404:
raise ObjectNotFound("Artist not found")
else:
self.request.map_json(request.json(), parse=self.parse_artist)

def parse_artist(self, json_obj: JsonObj) -> "Artist":
"""
"""Parses a TIDAL artist, replaces the current :class:`Artist` object. Made for
use within the python tidalapi module.
:param json_obj:
:return:
:param json_obj: :class:`JsonObj` containing the artist metadata
:return: Returns a copy of the :class:`Artist` object
"""
self.id = json_obj["id"]
self.name = json_obj["name"]
Expand All @@ -81,11 +91,11 @@ def parse_artist(self, json_obj: JsonObj) -> "Artist":
return copy.copy(self)

def parse_artists(self, json_obj: List[JsonObj]) -> List["Artist"]:
"""Parses a TIDAL artist, replaces the current artist object. Made for use
inside of the python tidalapi module.
"""Parses a list of TIDAL artists, returns a list of :class:`Artist` objects
Made for use within the python tidalapi module.
:param json_obj: Json data returned from api.tidal.com containing an artist
:return: Returns a copy of the original :exc: 'Artist': object
:param List[JsonObj] json_obj: List of :class:`JsonObj` containing the artist metadata for each artist
:return: Returns a list of :class:`Artist` objects
"""
return list(map(self.parse_artist, json_obj))

Expand Down
22 changes: 17 additions & 5 deletions tidalapi/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,21 +6,33 @@ class MediaUnknown(Exception):
pass


class UnknownManifestFormat(Exception):
class AssetNotAvailable(Exception):
pass


class URLNotAvailable(Exception):
pass


class MediaMissing(Exception):
class StreamNotAvailable(Exception):
pass


class StreamManifestDecodeError(Exception):
class MetadataNotAvailable(Exception):
pass


class ObjectNotFound(Exception):
pass


class UnknownManifestFormat(Exception):
pass


class MPDUnavailableError(Exception):
class ManifestDecodeError(Exception):
pass


class MPDDecodeError(Exception):
class MPDNotAvailableError(Exception):
pass
93 changes: 64 additions & 29 deletions tidalapi/media.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@

import copy
from abc import abstractmethod
from datetime import datetime
from datetime import datetime, timedelta
from enum import Enum
from typing import TYPE_CHECKING, List, Optional, Union, cast

Expand All @@ -36,16 +36,21 @@

import base64
import json
import re

import isodate
from isodate import parse_duration
from mpegdash.parser import MPEGDASHParser

from tidalapi.exceptions import *
from tidalapi.exceptions import (
AssetNotAvailable,
ManifestDecodeError,
MetadataNotAvailable,
MPDNotAvailableError,
StreamNotAvailable,
UnknownManifestFormat,
URLNotAvailable,
)
from tidalapi.types import JsonObj

# from mpd_parser.parser import Parser


class Quality(Enum):
low_96k = "LOW"
Expand Down Expand Up @@ -279,14 +284,24 @@ def _get(self, media_id: str) -> "Track":
:param media_id: TIDAL's identifier of the track
:return: A :class:`Track` object containing all the information about the track
"""
parse = self.parse_track
track = self.requests.map_request("tracks/%s" % media_id, parse=parse)
assert not isinstance(track, list)
return cast("Track", track)

request = self.requests.request("GET", "tracks/%s" % media_id)
if request.status_code and request.status_code == 404:
raise AssetNotAvailable("Track not available or not found")
else:
json_obj = request.json()
track = self.requests.map_json(json_obj, parse=self.parse_track)
assert not isinstance(track, list)
return cast("Track", track)

def get_url(self) -> str:
"""Retrieves the URL for a track.
:return: A `str` object containing the direct track URL
:raises: A :class:`exceptions.URLNotAvailable` if no URL is available for this track
"""
if self.session.is_pkce:
raise Exception(
raise URLNotAvailable(
"Track URL not available with quality:'{}'".format(
self.session.config.quality
)
Expand All @@ -296,25 +311,26 @@ def get_url(self) -> str:
"audioquality": self.session.config.quality,
"assetpresentation": "FULL",
}
json_obj = self.requests.map_request(
"tracks/%s/urlpostpaywall" % self.id, params
request = self.requests.request(
"GET", "tracks/%s/urlpostpaywall" % self.id, params
)
if json_obj.get("status") and json_obj.get("status") == 404:
raise AttributeError("URL not available for this track")
if request.status_code and request.status_code == 404:
raise URLNotAvailable("URL not available for this track")
else:
json_obj = request.json()
return cast(str, json_obj["urls"][0])

def lyrics(self) -> "Lyrics":
"""Retrieves the lyrics for a song.
:return: A :class:`Lyrics` object containing the lyrics
:raises: A :class:`requests.HTTPError` if there aren't any lyrics
:raises: A :class:`exceptions.MetadataNotAvailable` if there aren't any lyrics
"""

json_obj = self.requests.map_request("tracks/%s/lyrics" % self.id)
if json_obj.get("status") and json_obj.get("status") == 404:
raise AttributeError("No lyrics exists for this track")
request = self.requests.request("GET", "tracks/%s/lyrics" % self.id)
if request.status_code and request.status_code == 404:
raise MetadataNotAvailable("No lyrics exists for this track")
else:
json_obj = request.json()
lyrics = self.requests.map_json(json_obj, parse=Lyrics().parse)
assert not isinstance(lyrics, list)
return cast("Lyrics", lyrics)
Expand All @@ -324,32 +340,41 @@ def get_track_radio(self, limit: int = 100) -> List["Track"]:
to this track.
:return: A list of :class:`Tracks <tidalapi.media.Track>`
:raises: A :class:`exceptions.MetadataNotAvailable` if no track radio is available
"""
params = {"limit": limit}
tracks = self.requests.map_request(
"tracks/%s/radio" % self.id, params=params, parse=self.session.parse_track

request = self.requests.request(
"GET", "tracks/%s/radio" % self.id, params=params
)
assert isinstance(tracks, list)
return cast(List["Track"], tracks)
if request.status_code and request.status_code == 404:
raise MetadataNotAvailable("Track radio not available for this track")
else:
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)

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
parameters needed for streaming via `MPEG-DASH` protocol.
:raises: A :class:`exceptions.StreamNotAvailable` if there is no stream available for this track
"""
params = {
"playbackmode": "STREAM",
"audioquality": self.session.config.quality,
"assetpresentation": "FULL",
}

json_obj = self.requests.map_request(
"tracks/%s/playbackinfopostpaywall" % self.id, params
request = self.requests.request(
"GET", "tracks/%s/playbackinfopostpaywall" % self.id, params
)
if json_obj.get("status") and json_obj.get("status") == 404:
raise AttributeError("Stream not available for this track")
if request.status_code and request.status_code == 404:
raise StreamNotAvailable("Stream not available for this track")
else:
json_obj = request.json()
stream = self.requests.map_json(json_obj, parse=Stream().parse)
assert not isinstance(stream, list)
return cast("Stream", stream)
Expand Down Expand Up @@ -697,15 +722,25 @@ def _get(self, media_id: str) -> Video:
return cast("Video", video)

def get_url(self) -> str:
"""Retrieves the URL for a video.
:return: A `str` object containing the direct video URL
:raises: A :class:`exceptions.URLNotAvailable` if no URL is available for this video
"""
params = {
"urlusagemode": "STREAM",
"videoquality": self.session.config.video_quality,
"assetpresentation": "FULL",
}

request = self.requests.request(
"GET", "videos/%s/urlpostpaywall" % self.id, params
)
return cast(str, request.json()["urls"][0])
if request.status_code and request.status_code == 404:
raise URLNotAvailable("URL not available for this video")
else:
json_obj = request.json()
return cast(str, json_obj["urls"][0])

def image(self, width: int = 1080, height: int = 720) -> str:
if (width, height) not in [(160, 107), (480, 320), (750, 500), (1080, 720)]:
Expand Down
10 changes: 7 additions & 3 deletions tidalapi/playlist.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
from datetime import datetime
from typing import TYPE_CHECKING, List, Optional, Sequence, Union, cast

from tidalapi.exceptions import ObjectNotFound
from tidalapi.types import JsonObj
from tidalapi.user import LoggedInUser

Expand Down Expand Up @@ -65,9 +66,12 @@ def __init__(self, session: "Session", playlist_id: Optional[str]):
self.requests = session.request
self._base_url = "playlists/%s"
if playlist_id:
request = self.requests.request("GET", self._base_url % playlist_id)
self._etag = request.headers["etag"]
self.parse(request.json())
request = self.requests.request("GET", self._base_url % self.id)
if request.status_code and request.status_code == 404:
raise ObjectNotFound("Playlist not found")
else:
self._etag = request.headers["etag"]
self.parse(request.json())

def parse(self, json_obj: JsonObj) -> "Playlist":
"""Parses a playlist from tidal, replaces the current playlist object.
Expand Down

0 comments on commit e907654

Please sign in to comment.