Centralized error handling and added after_retry to TooManyRequests error.

HISTORY.rst

@@ -2,6 +2,12 @@
 
 History
 =======
+
+Upcoming
+--------
+* TooManyRequests now includes the retry_after header in its data. - semohr_
+* Added a central error class (TidalAPIError) to allow for unified error handling. - semohr_
+
 v0.8.6
 ------
 * Add support for get<track, album, artist, playlist>count(), Workers: Use get_*_count to get the actual number of items. - tehkillerbee_
@@ -242,6 +248,7 @@ v0.6.2
 * Add version tag for Track - Husky22_
 * Switch to netlify for documentation - morguldir_
 
+.. _semohr: https://github.com/semohr
 .. _morguldir: https://github.com/morguldir
 .. _Husky22: https://github.com/Husky22
 .. _ktnrg45: https://github.com/ktnrg45

tidalapi/album.py

@@ -87,10 +87,11 @@ def __init__(self, session: "Session", album_id: Optional[str]):
         if self.id:
             try:
                 request = self.request.request("GET", "albums/%s" % self.id)
-            except ObjectNotFound:
-                raise ObjectNotFound("Album not found")
-            except TooManyRequests:
-                raise TooManyRequests("Album unavailable")
+            except ObjectNotFound as e:
+                e.args = ("Album with id %s not found" % self.id,)
+            except TooManyRequests as e:
+                e.args = ("Album unavailable",)
+                raise e
             else:
                 self.request.map_json(request.json(), parse=self.parse)
 
@@ -320,8 +321,9 @@ def similar(self) -> List["Album"]:
             request = self.request.request("GET", "albums/%s/similar" % self.id)
         except ObjectNotFound:
             raise MetadataNotAvailable("No similar albums exist for this album")
-        except TooManyRequests:
-            raise TooManyRequests("Similar artists unavailable")
+        except TooManyRequests as e:
+            e.args = ("Similar artists unavailable",)
+            raise e
         else:
             albums = self.request.map_json(
                 request.json(), parse=self.session.parse_album

tidalapi/artist.py

@@ -64,10 +64,12 @@ def __init__(self, session: "Session", artist_id: Optional[str]):
         if self.id:
             try:
                 request = self.request.request("GET", "artists/%s" % self.id)
-            except ObjectNotFound:
-                raise ObjectNotFound("Artist not found")
-            except TooManyRequests:
-                raise TooManyRequests("Artist unavailable")
+            except ObjectNotFound as e:
+                e.args = ("Artist with id %s not found" % self.id,)
+                raise e
+            except TooManyRequests as e:
+                e.args = ("Artist unavailable",)
+                raise e
             else:
                 self.request.map_json(request.json(), parse=self.parse_artist)
 
@@ -242,8 +244,9 @@ def get_radio(self, limit: int = 100) -> List["Track"]:
             )
         except ObjectNotFound:
             raise MetadataNotAvailable("Track radio not available for this track")
-        except TooManyRequests:
-            raise TooManyRequests("Track radio unavailable")
+        except TooManyRequests as e:
+            e.args = ("Track radio unavailable",)
+            raise e
         else:
             json_obj = request.json()
             radio = self.request.map_json(json_obj, parse=self.session.parse_track)
@@ -262,8 +265,9 @@ def get_radio_mix(self) -> mix.Mix:
             request = self.request.request("GET", "artists/%s/mix" % self.id)
         except ObjectNotFound:
             raise MetadataNotAvailable("Artist radio not available for this artist")
-        except TooManyRequests:
-            raise TooManyRequests("Artist radio unavailable")
+        except TooManyRequests as e:
+            e.args = ("Artist radio unavailable",)
+            raise e
         else:
             json_obj = request.json()
             return self.session.mix(json_obj.get("id"))

tidalapi/exceptions.py

@@ -1,46 +1,86 @@
-class AuthenticationError(Exception):
+from __future__ import annotations
+
+import json
+import logging
+
+from requests import HTTPError
+
+log = logging.getLogger(__name__)
+
+
+class TidalAPIError(Exception):
     pass
 
 
-class AssetNotAvailable(Exception):
+class AuthenticationError(TidalAPIError):
     pass
 
 
-class TooManyRequests(Exception):
+class AssetNotAvailable(TidalAPIError):
     pass
 
 
-class URLNotAvailable(Exception):
+class TooManyRequests(TidalAPIError):
+    retry_after: int
+
+    def __init__(self, message: str = "Too many requests", retry_after: int = -1):
+        super().__init__(message)
+        self.retry_after = retry_after
+
+
+class URLNotAvailable(TidalAPIError):
     pass
 
 
-class StreamNotAvailable(Exception):
+class StreamNotAvailable(TidalAPIError):
     pass
 
 
-class MetadataNotAvailable(Exception):
+class MetadataNotAvailable(TidalAPIError):
     pass
 
 
-class ObjectNotFound(Exception):
+class ObjectNotFound(TidalAPIError):
     pass
 
 
-class UnknownManifestFormat(Exception):
+class UnknownManifestFormat(TidalAPIError):
     pass
 
 
-class ManifestDecodeError(Exception):
+class ManifestDecodeError(TidalAPIError):
     pass
 
 
-class MPDNotAvailableError(Exception):
+class MPDNotAvailableError(TidalAPIError):
     pass
 
 
-class InvalidISRC(Exception):
+class InvalidISRC(TidalAPIError):
     pass
 
 
-class InvalidUPC(Exception):
+class InvalidUPC(TidalAPIError):
     pass
+
+
+def http_error_to_tidal_error(http_error: HTTPError) -> TidalAPIError | None:
+    response = http_error.response
+
+    if response.content:
+        json_data = response.json()
+        # Make sure request response contains the detailed error message
+        if "errors" in json_data:
+            log.debug("Request response: '%s'", json_data["errors"][0]["detail"])
+        elif "userMessage" in json_data:
+            log.debug("Request response: '%s'", json_data["userMessage"])
+        else:
+            log.debug("Request response: '%s'", json.dumps(json_data))
+
+    elif response.status_code == 404:
+        return ObjectNotFound("Object not found")
+    elif response.status_code == 429:
+        retry_after = int(response.headers.get("Retry-After", -1))
+        return TooManyRequests("Too many requests", retry_after=retry_after)
+
+    return None

tidalapi/media.py

@@ -329,10 +329,12 @@ def _get(self, media_id: str) -> "Track":
 
         try:
             request = self.requests.request("GET", "tracks/%s" % media_id)
-        except ObjectNotFound:
-            raise ObjectNotFound("Track not found or unavailable")
-        except TooManyRequests:
-            raise TooManyRequests("Track unavailable")
+        except ObjectNotFound as e:
+            e.args = ("Track with id %s not found" % media_id,)
+            raise e
+        except TooManyRequests as e:
+            e.args = ("Track unavailable",)
+            raise e
         else:
             json_obj = request.json()
             track = self.requests.map_json(json_obj, parse=self.parse_track)
@@ -362,8 +364,9 @@ def get_url(self) -> str:
             )
         except ObjectNotFound:
             raise URLNotAvailable("URL not available for this track")
-        except TooManyRequests:
-            raise TooManyRequests("URL Unavailable")
+        except TooManyRequests as e:
+            e.args = ("URL unavailable",)
+            raise e
         else:
             json_obj = request.json()
             return cast(str, json_obj["urls"][0])
@@ -378,8 +381,9 @@ def lyrics(self) -> "Lyrics":
             request = self.requests.request("GET", "tracks/%s/lyrics" % self.id)
         except ObjectNotFound:
             raise MetadataNotAvailable("No lyrics exists for this track")
-        except TooManyRequests:
-            raise TooManyRequests("Lyrics unavailable")
+        except TooManyRequests as e:
+            e.args = ("Lyrics unavailable",)
+            raise e
         else:
             json_obj = request.json()
             lyrics = self.requests.map_json(json_obj, parse=Lyrics().parse)
@@ -401,8 +405,9 @@ def get_track_radio(self, limit: int = 100) -> List["Track"]:
             )
         except ObjectNotFound:
             raise MetadataNotAvailable("Track radio not available for this track")
-        except TooManyRequests:
-            raise TooManyRequests("Track radio unavailable")
+        except TooManyRequests as e:
+            e.args = ("Track radio unavailable",)
+            raise e
         else:
             json_obj = request.json()
             tracks = self.requests.map_json(json_obj, parse=self.session.parse_track)
@@ -420,8 +425,9 @@ def get_radio_mix(self) -> mix.Mix:
             request = self.requests.request("GET", "tracks/%s/mix" % self.id)
         except ObjectNotFound:
             raise MetadataNotAvailable("Track radio not available for this track")
-        except TooManyRequests:
-            raise TooManyRequests("Track radio unavailable")
+        except TooManyRequests as e:
+            e.args = ("Track radio unavailable",)
+            raise e
         else:
             json_obj = request.json()
             return self.session.mix(json_obj.get("id"))
@@ -445,8 +451,9 @@ def get_stream(self) -> "Stream":
             )
         except ObjectNotFound:
             raise StreamNotAvailable("Stream not available for this track")
-        except TooManyRequests:
-            raise TooManyRequests("Stream unavailable")
+        except TooManyRequests as e:
+            e.args = ("Stream unavailable",)
+            raise e
         else:
             json_obj = request.json()
             stream = self.requests.map_json(json_obj, parse=Stream().parse)
@@ -863,10 +870,12 @@ def _get(self, media_id: str) -> Video:
 
         try:
             request = self.requests.request("GET", "videos/%s" % self.id)
-        except ObjectNotFound:
-            raise ObjectNotFound("Video not found or unavailable")
-        except TooManyRequests:
-            raise TooManyRequests("Video unavailable")
+        except ObjectNotFound as e:
+            e.args = ("Video with id %s not found" % media_id,)
+            raise e
+        except TooManyRequests as e:
+            e.args = ("Video unavailable",)
+            raise e
         else:
             json_obj = request.json()
             video = self.requests.map_json(json_obj, parse=self.parse_video)
@@ -891,8 +900,9 @@ def get_url(self) -> str:
             )
         except ObjectNotFound:
             raise URLNotAvailable("URL not available for this video")
-        except TooManyRequests:
-            raise TooManyRequests("URL unavailable)")
+        except TooManyRequests as e:
+            e.args = ("URL unavailable",)
+            raise e
         else:
             json_obj = request.json()
             return cast(str, json_obj["urls"][0])

tidalapi/mix.py

@@ -95,10 +95,12 @@ def get(self, mix_id: Optional[str] = None) -> "Mix":
 
         try:
             request = self.request.request("GET", "pages/mix", params=params)
-        except ObjectNotFound:
-            raise ObjectNotFound("Mix not found")
-        except TooManyRequests:
-            raise TooManyRequests("Mix unavailable")
+        except ObjectNotFound as e:
+            e.args = ("Mix with id %s not found" % mix_id,)
+            raise e
+        except TooManyRequests as e:
+            e.args = ("Mix unavailable",)
+            raise e
         else:
             result = self.session.parse_page(request.json())
             assert not isinstance(result, list)
@@ -215,10 +217,12 @@ def get(self, mix_id: Optional[str] = None) -> "MixV2":
         params = {"mixId": mix_id, "deviceType": "BROWSER"}
         try:
             request = self.request.request("GET", "pages/mix", params=params)
-        except ObjectNotFound:
-            raise ObjectNotFound("Mix not found")
-        except TooManyRequests:
-            raise TooManyRequests("Mix unavailable")
+        except ObjectNotFound as e:
+            e.args = ("Mix with id %s not found" % mix_id,)
+            raise e
+        except TooManyRequests as e:
+            e.args = ("Mix unavailable",)
+            raise e
         else:
             result = self.session.parse_page(request.json())
             assert not isinstance(result, list)

tidalapi/playlist.py

@@ -81,10 +81,12 @@ def __init__(self, session: "Session", playlist_id: Optional[str]):
         if playlist_id:
             try:
                 request = self.request.request("GET", self._base_url % self.id)
-            except ObjectNotFound:
-                raise ObjectNotFound("Playlist not found")
-            except TooManyRequests:
-                raise TooManyRequests("Playlist unavailable")
+            except ObjectNotFound as e:
+                e.args = ("Playlist with id %s not found" % playlist_id,)
+                raise e
+            except TooManyRequests as e:
+                e.args = ("Playlist unavailable",)
+                raise e
             else:
                 self._etag = request.headers["etag"]
                 self.parse(request.json())
@@ -370,9 +372,10 @@ def __init__(
                         return
                 raise ObjectNotFound
             except ObjectNotFound:
-                raise ObjectNotFound(f"Folder not found")
-            except TooManyRequests:
-                raise TooManyRequests("Folder unavailable")
+                raise ObjectNotFound("Folder not found")
+            except TooManyRequests as e:
+                e.args = ("Folder unavailable",)
+                raise e
 
     def _reparse(self) -> None:
         params = {

tidalapi/request.py

@@ -34,7 +34,7 @@
 
 import requests
 
-from tidalapi.exceptions import ObjectNotFound, TooManyRequests
+from tidalapi.exceptions import http_error_to_tidal_error
 from tidalapi.types import JsonObj
 
 log = logging.getLogger(__name__)
@@ -151,26 +151,13 @@ def request(
         log.debug("request: %s", request.request.url)
         try:
             request.raise_for_status()
-        except Exception as e:
+        except requests.HTTPError as e:
             log.info("Request resulted in exception {}".format(e))
             self.latest_err_response = request
-            if request.content:
-                resp = request.json()
-                # Make sure request response contains the detailed error message
-                if "errors" in resp:
-                    log.debug("Request response: '%s'", resp["errors"][0]["detail"])
-                elif "userMessage" in resp:
-                    log.debug("Request response: '%s'", resp["userMessage"])
-                else:
-                    log.debug("Request response: '%s'", json.dumps(resp))
-
-            if request.status_code and request.status_code == 404:
-                raise ObjectNotFound
-            elif request.status_code and request.status_code == 429:
-                raise TooManyRequests
+            if err := http_error_to_tidal_error(e):
+                raise err from e
             else:
-                # raise last error, usually HTTPError
-                raise
+                raise  # re raise last error, usually HTTPError
         return request
 
     def get_latest_err_response(self) -> dict: