Merge pull request #276 from tamland/filter-isrc-and-barcode-support

Filter isrc and barcode support

HISTORY.rst

@@ -2,6 +2,13 @@
 
 History
 =======
+v0.7.7
+------
+* Add method to get detailed request error response if an error occurred during request. - tehkillerbee_
+* Tests: Add tests tests for ISRC, barcode methods and cleanup exception handling. - tehkillerbee_
+* Feat.: Add support to get tracks by ISRC. - tehkillerbee_, M4TH1EU_
+* Feat.: Add support to get albums by Barcode ID (UPC). - tehkillerbee_, M4TH1EU_
+* Feat.: Add support for a custom base url in `request()` and `basic_request()` to use the new openapi. - M4TH1EU_
 
 v0.7.6
 ------
@@ -178,6 +185,7 @@ v0.6.2
 .. _arnesongit: https://github.com/arnesongit
 .. _Jimmyscene: https://github.com/Jimmyscene
 .. _quodrum-glas: https://github.com/quodrum-glas
+.. _M4TH1EU: https://github.com/M4TH1EU
 
 
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "tidalapi"
-version = "0.7.6"
+version = "0.7.7"
 description = "Unofficial API for TIDAL music streaming service."
 authors = ["Thomas Amland <thomas.amland@googlemail.com>"]
 maintainers = ["tehkillerbee <tehkillerbee@users.noreply.github.com>"]

tests/test_session.py

@@ -24,6 +24,7 @@
 
 import tidalapi
 from tidalapi import Album, Artist, Playlist, Track, Video
+from tidalapi.exceptions import InvalidISRC, InvalidUPC, ObjectNotFound
 
 
 def test_load_oauth_session(session):
@@ -161,3 +162,27 @@ def test_manually_set_video_quality_is_preserved(session, quality):
     session.video_quality = quality
     assert session.video_quality == quality
     assert session.config.video_quality == quality
+
+
+def test_tracks_by_isrc(session):
+    # Track found using USSM12209515 => track id 251380837
+    tracks = session.get_tracks_by_isrc("USSM12209515")
+    assert tracks[0].id == 251380837
+    # ISRC valid but track not found (ObjectNotFound)
+    with pytest.raises(ObjectNotFound):
+        session.get_tracks_by_isrc("QMEU32403189")
+    # Invalid isrc (InvalidISRC)
+    with pytest.raises(InvalidISRC):
+        session.get_tracks_by_isrc("12209515")
+
+
+def test_albums_by_barcode(session):
+    # Track found using barcode 196589525444 => album id 251380836
+    albums = session.get_albums_by_barcode("196589525444")
+    assert albums[0].id == 251380836
+    # ISRC valid but track not found (ObjectNotFound)
+    with pytest.raises(ObjectNotFound):
+        session.get_albums_by_barcode("112233445566")
+    # Invalid Barcode UPC (InvalidUPC)
+    with pytest.raises(InvalidUPC):
+        session.get_albums_by_barcode("aaaa")

tidalapi/__init__.py

@@ -19,4 +19,4 @@
     User,
 )
 
-__version__ = "0.7.6"
+__version__ = "0.7.7"

tidalapi/exceptions.py

@@ -36,3 +36,11 @@ class ManifestDecodeError(Exception):
 
 class MPDNotAvailableError(Exception):
     pass
+
+
+class InvalidISRC(Exception):
+    pass
+
+
+class InvalidUPC(Exception):
+    pass

tidalapi/request.py

@@ -53,12 +53,15 @@ class Requests(object):
     """A class for handling api requests to TIDAL."""
 
     user_agent: str
+    # Latest error response that can be returned and parsed after request has been completed
+    latest_err_response: requests.Response
 
     def __init__(self, session: "Session"):
         # More Android User-Agents here: https://user-agents.net/browsers/android
         self.user_agent = "Mozilla/5.0 (Linux; Android 12; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/91.0.4472.114 Safari/537.36"
         self.session = session
         self.config = session.config
+        self.latest_err_response = requests.Response()
 
     def basic_request(
         self,
@@ -67,6 +70,7 @@ def basic_request(
         params: Optional[Params] = None,
         data: Optional[JsonObj] = None,
         headers: Optional[MutableMapping[str, str]] = None,
+        base_url: Optional[str] = None,
     ) -> requests.Response:
         request_params = {
             "sessionId": self.session.session_id,
@@ -90,7 +94,10 @@ def basic_request(
             headers["authorization"] = (
                 self.session.token_type + " " + self.session.access_token
             )
-        url = urljoin(self.session.config.api_v1_location, path)
+        if base_url is None:
+            base_url = self.session.config.api_v1_location
+
+        url = urljoin(base_url, path)
         request = self.session.request_session.request(
             method, url, params=request_params, data=data, headers=headers
         )
@@ -111,7 +118,7 @@ def basic_request(
                 if refreshed:
                     request = self.basic_request(method, url, params, data, headers)
             else:
-                log.warning("HTTP error on %d", request.status_code)
+                log.debug("HTTP error on %d", request.status_code)
                 log.debug("Response text\n%s", request.text)
 
         return request
@@ -123,6 +130,7 @@ def request(
         params: Optional[Params] = None,
         data: Optional[JsonObj] = None,
         headers: Optional[MutableMapping[str, str]] = None,
+        base_url: Optional[str] = None,
     ) -> requests.Response:
         """Method for tidal requests.
 
@@ -133,18 +141,20 @@ def request(
         :param params: The parameters you want to supply with the request.
         :param data: The data you want to supply with the request.
         :param headers: The headers you want to include with the request
+        :param base_url: The base url to use for the request
         :return: The json data at specified api endpoint.
         """
 
-        request = self.basic_request(method, path, params, data, headers)
+        request = self.basic_request(method, path, params, data, headers, base_url)
         log.debug("request: %s", request.request.url)
         try:
             request.raise_for_status()
         except Exception as e:
-            log.info("Got exception {}".format(e))
-            log.debug("Response was {}".format(e.response))
+            log.info("Request resulted in exception {}".format(e))
+            self.latest_err_response = request
             if request.content:
-                log.debug("response: %s", json.dumps(request.json(), indent=4))
+                resp = request.json()
+                log.debug("Request response: '%s'", resp["errors"][0]["detail"])
             if request.status_code and request.status_code == 404:
                 raise ObjectNotFound
             elif request.status_code and request.status_code == 429:
@@ -154,6 +164,25 @@ def request(
                 raise
         return request
 
+    def get_latest_err_response(self) -> dict:
+        """Get the latest request Response that resulted in an Exception :return: The
+        request Response that resulted in the Exception, returned as a dict An empty
+        dict will be returned, if no response was returned."""
+        if self.latest_err_response.content:
+            return self.latest_err_response.json()
+        else:
+            return {}
+
+    def get_latest_err_response_str(self) -> str:
+        """Get the latest request response message as a string :return: The contents of
+        the (detailed) error response Response, returned as a string An empty str will
+        be returned, if no response was returned."""
+        if self.latest_err_response.content:
+            resp = self.latest_err_response.json()
+            return resp["errors"][0]["detail"]
+        else:
+            return ""
+
     def map_request(
         self,
         url: str,

tidalapi/session.py

@@ -49,6 +49,7 @@
 from urllib.parse import parse_qs, urlencode, urljoin, urlsplit
 
 import requests
+from requests.exceptions import HTTPError
 
 from tidalapi.exceptions import *
 from tidalapi.types import JsonObj
@@ -103,6 +104,7 @@ class Config:
     api_pkce_auth: str = "https://login.tidal.com/authorize"
     api_v1_location: str = "https://api.tidal.com/v1/"
     api_v2_location: str = "https://api.tidal.com/v2/"
+    openapi_v2_location: str = "https://openapi.tidal.com/v2/"
     api_token: str
     client_id: str
     client_secret: str
@@ -857,6 +859,70 @@ def track(
             log.warning("Track '%s' is unavailable", track_id)
             raise
 
+    def get_tracks_by_isrc(self, isrc: str) -> list[media.Track]:
+        """Function to search all tracks with a specific ISRC code. (eg. "USSM12209515")
+        This method uses the TIDAL openapi (v2). See the apiref below for more details:
+        https://apiref.developer.tidal.com/apiref?spec=catalogue-v2&ref=get-tracks-v2
+
+        :param isrc: The ISRC of the Track.
+        :return: Returns a list of :class:`.Track` objects that have access to the session instance used.
+                 An empty list will be returned if no tracks matches the ISRC
+        """
+        try:
+            params = {
+                "filter[isrc]": isrc,
+            }
+            res = self.request.request(
+                "GET",
+                "tracks",
+                params=params,
+                base_url=self.config.openapi_v2_location,
+            ).json()
+            if res["data"]:
+                return [self.track(tr["id"]) for tr in res["data"]]
+            else:
+                log.warning("No matching tracks found for ISRC '%s'", isrc)
+                raise ObjectNotFound
+        except HTTPError:
+            log.error("Invalid ISRC code '%s'", isrc)
+            # Get latest detailed error response and return the response given from the TIDAL api
+            resp_str = self.request.get_latest_err_response_str()
+            if resp_str:
+                log.error("API Response: '%s'", resp_str)
+            raise InvalidISRC
+
+    def get_albums_by_barcode(self, barcode: str) -> list[album.Album]:
+        """Function to search all albums with a specific UPC code (eg. "196589525444")
+        This method uses the TIDAL openapi (v2). See the apiref below for more details:
+        https://apiref.developer.tidal.com/apiref?spec=catalogue-v2&ref=get-albums-v2
+
+        :param barcode: The UPC of the Album. Eg.
+        :return: Returns a list of :class:`.Album` objects that have access to the session instance used.
+                 An empty list will be returned if no tracks matches the ISRC
+        """
+        try:
+            params = {
+                "filter[barcodeId]": barcode,
+            }
+            res = self.request.request(
+                "GET",
+                "albums",
+                params=params,
+                base_url=self.config.openapi_v2_location,
+            ).json()
+            if res["data"]:
+                return [self.album(alb["id"]) for alb in res["data"]]
+            else:
+                log.warning("No matching albums found for UPC barcode '%s'", barcode)
+                raise ObjectNotFound
+        except HTTPError:
+            log.error("Invalid UPC barcode '%s'.", barcode)
+            # Get latest detailed error response and return the response given from the TIDAL api
+            resp_str = self.request.get_latest_err_response_str()
+            if resp_str:
+                log.error("API Response: '%s'", resp_str)
+            raise InvalidUPC
+
     def video(self, video_id: Optional[str] = None) -> media.Video:
         """Function to create a Video object with access to the session instance in a
         smoother way. Calls :class:`tidalapi.Video(session=session, video_id=video_id)

tidalapi/user.py

@@ -394,7 +394,9 @@ def mixes(self, limit: Optional[int] = 50, offset: int = 0) -> List["MixV2"]:
         return cast(
             List["MixV2"],
             self.requests.map_request(
-                url=urljoin("https://api.tidal.com/v2/", f"{self.v2_base_url}/mixes"),
+                url=urljoin(
+                    self.session.config.api_v2_location, f"{self.v2_base_url}/mixes"
+                ),
                 params=params,
                 parse=self.session.parse_v2_mix,
             ),