Make it faster

pyproject.toml

@@ -31,11 +31,11 @@ dependencies = [
     "backoff==2.2.1",
     "dataclass-wizard==0.22.3",
     "ffmpeg-python==0.2.0",
+    "httpx[brotli,http2]==0.27.0",
     "mutagen==1.47.0",
     "m3u8==4.0.0",
     "platformdirs==4.2.0",
     "pycryptodome==3.20.0",
-    "requests==2.31.0",
     "typer==0.9.0"
 ]
 [project.optional-dependencies]

requirements.txt

@@ -1,9 +1,9 @@
 backoff==2.2.1
 dataclass-wizard==0.22.3
 ffmpeg-python==0.2.0
+httpx[http2,brotli]==0.27.0
 mutagen==1.47.0
 m3u8==4.0.0
 platformdirs==4.2.0
 pycryptodome==3.20.0
-requests==2.31.0
 typer==0.9.0

tidal_wave/album.py

@@ -5,8 +5,6 @@
 import sys
 from typing import List, Optional, Tuple
 
-from requests import Session
-
 from .media import AudioFormat
 from .models import (
     AlbumsCreditsResponseJSON,
@@ -24,6 +22,8 @@
 from .track import Track
 from .utils import download_cover_image
 
+from httpx import Client
+
 logger = logging.getLogger("__name__")
 
 
@@ -35,43 +35,43 @@ def __post_init__(self):
         self.album_dir: Optional[Path] = None
         self.album_cover_saved: bool = False
 
-    def set_tracks(self, session: Session):
+    def set_tracks(self, client: Client):
         """This method populates self.tracks by requesting from
         TIDAL albums/items endpoint."""
         album_items: AlbumsItemsResponseJSON = request_album_items(
-            session=session, album_id=self.album_id
+            client=client, album_id=self.album_id
         )
         _items = album_items.items if album_items is not None else ()
         self.tracks: Tuple[TracksEndpointResponseJSON] = tuple(
             _item.item for _item in _items
         )
 
-    def set_metadata(self, session: Session):
+    def set_metadata(self, client: Client):
         """This method sets self.metadata by requesting from
         TIDAL /albums endpoint"""
         self.metadata: AlbumsEndpointResponseJSON = request_albums(
-            session=session, album_id=self.album_id
+            client=client, album_id=self.album_id
         )
 
-    def set_album_review(self, session: Session):
+    def set_album_review(self, client: Client):
         """This method requests the review corresponding to self.album_id
         in TIDAL. If it exists, it is written to disk as AlbumReview.json
         in self.album_dir"""
         self.album_review: Optional[AlbumsReviewResponseJSON] = request_album_review(
-            session=session, album_id=self.album_id
+            client=client, album_id=self.album_id
         )
         if self.album_review is not None:
             (self.album_dir / "AlbumReview.json").write_text(
                 self.album_review.to_json()
             )
 
-    def set_album_credits(self, session: Session):
+    def set_album_credits(self, client: Client):
         """This method requests the album's top-level credits (separate from
         each track's credits) and writes them to AlbumCredits.json in
         self.album_dir"""
-        self.album_credits: Optional[AlbumsCreditsResponseJSON] = (
-            request_albums_credits(session=session, album_id=self.album_id)
-        )
+        self.album_credits: Optional[
+            AlbumsCreditsResponseJSON
+        ] = request_albums_credits(client=client, album_id=self.album_id)
         if self.album_credits is not None:
             f: str = str((self.album_dir / "AlbumCredits.json").absolute())
             with open(f, "w") as fp:
@@ -96,7 +96,7 @@ def set_album_dir(self, out_dir: Path):
                     parents=True, exist_ok=True
                 )
 
-    def save_cover_image(self, session: Session, out_dir: Path):
+    def save_cover_image(self, client: Client, out_dir: Path):
         """This method writes cover.jpg in self.album_dir via the
         utils.download_cover_image() function. If successful,
         then self.album_cover_saved takes the value True"""
@@ -105,7 +105,7 @@ def save_cover_image(self, session: Session, out_dir: Path):
         self.cover_path: Path = self.album_dir / "cover.jpg"
         if not self.cover_path.exists():
             download_cover_image(
-                session=session,
+                client=client,
                 cover_uuid=self.metadata.cover,
                 output_dir=self.album_dir,
             )
@@ -114,7 +114,7 @@ def save_cover_image(self, session: Session, out_dir: Path):
 
     def get_tracks(
         self,
-        session: Session,
+        client: Client,
         audio_format: AudioFormat,
         out_dir: Path,
         no_extra_files: bool,
@@ -127,7 +127,7 @@ def get_tracks(
             track: Track = Track(track_id=t.id)
 
             track_files_value: Optional[str] = track.get(
-                session=session,
+                client=client,
                 audio_format=audio_format,
                 out_dir=out_dir,
                 metadata=t,
@@ -149,7 +149,7 @@ def dump(self, fp=sys.stdout):
 
     def get(
         self,
-        session: Session,
+        client: Client,
         audio_format: AudioFormat,
         out_dir: Path,
         metadata: Optional[AlbumsEndpointResponseJSON] = None,
@@ -164,32 +164,32 @@ def get(
             5. get_tracks()
         """
         if metadata is None:
-            self.set_metadata(session)
+            self.set_metadata(client=client)
         else:
             self.metadata = metadata
 
         if self.metadata is None:
             self.track_files = {}
             return
 
-        self.set_tracks(session)
+        self.set_tracks(client=client)
 
-        self.set_album_dir(out_dir)
+        self.set_album_dir(out_dir=out_dir)
 
         if self.metadata.cover != "":  # None was sent from the API
-            self.save_cover_image(session, out_dir)
+            self.save_cover_image(client=client, out_dir=out_dir)
         else:
             logger.warning(
                 f"No cover image was returned from TIDAL API for album {self.album_id}"
             )
 
         if not no_extra_files:
-            self.set_album_review(session)
-            self.set_album_credits(session)
+            self.set_album_review(client)
+            self.set_album_credits(client)
         else:
             try:
                 self.cover_path.unlink()
             except FileNotFoundError:
                 pass
 
-        self.get_tracks(session, audio_format, out_dir, no_extra_files)
+        self.get_tracks(client, audio_format, out_dir, no_extra_files)

tidal_wave/artist.py

@@ -3,8 +3,6 @@
 from pathlib import Path
 from typing import List, Optional
 
-from requests import Session
-
 from .album import Album
 from .media import AudioFormat
 from .models import (
@@ -21,49 +19,51 @@
 from .utils import download_cover_image
 from .video import Video
 
+from httpx import Client
+
 logger = logging.getLogger("__name__")
 
 
 @dataclass
 class Artist:
     artist_id: int
 
-    def set_metadata(self, session: Session):
+    def set_metadata(self, client: Client):
         """This function requests from TIDAL API endpoint /artists and
         stores the results in self.metadata"""
         self.metadata: Optional[ArtistsEndpointResponseJSON] = request_artists(
-            session=session, artist_id=self.artist_id
+            client=client, artist_id=self.artist_id
         )
 
-    def save_artist_image(self, session: Session):
+    def save_artist_image(self, client: Client):
         """This method writes the bytes of self.metadata.picture to
         the file cover.jpg in self.artist_dir"""
         artist_image: Path = self.artist_dir / "cover.jpg"
         if not artist_image.exists():
             if self.metadata.picture is not None:
                 download_cover_image(
-                    session, self.metadata.picture, self.artist_dir, dimension=750
+                    client, self.metadata.picture, self.artist_dir, dimension=750
                 )
 
-    def set_albums(self, session: Session):
+    def set_albums(self, client: Client):
         """This method requests from TIDAL API endpoint /artists/albums and
         stores the results in self.albums"""
         self.albums: Optional[ArtistsAlbumsResponseJSON] = request_artists_albums(
-            session=session, artist_id=self.artist_id
+            client=client, artist_id=self.artist_id
         )
 
-    def set_audio_works(self, session: Session):
+    def set_audio_works(self, client: Client):
         """This method requests from TIDAL API endpoint
         /artists/albums?filter=EPSANDSINGLES and stores the results in self.albums"""
         self.albums: Optional[ArtistsAlbumsResponseJSON] = request_artists_audio_works(
-            session=session, artist_id=self.artist_id
+            client=client, artist_id=self.artist_id
         )
 
-    def set_videos(self, session: Session):
+    def set_videos(self, client: Client):
         """This method requests from TIDAL API endpoint /artists/videos and
         stores the results in self.albums"""
         self.videos: Optional[ArtistsVideosResponseJSON] = request_artists_videos(
-            session=session, artist_id=self.artist_id
+            client=client, artist_id=self.artist_id
         )
 
     def set_artist_dir(self, out_dir: Path):
@@ -75,7 +75,7 @@ def set_artist_dir(self, out_dir: Path):
 
     def get_albums(
         self,
-        session: Session,
+        client: Client,
         audio_format: AudioFormat,
         out_dir: Path,
         include_eps_singles: bool,
@@ -86,14 +86,14 @@ def get_albums(
         the albums (and, optionally, EPs and singles) is requested and
         written to subdirectories of out_dir"""
         if include_eps_singles:
-            self.set_audio_works(session)
+            self.set_audio_works(client)
             logger.info(
                 f"Starting attempt to get {self.albums.total_number_of_items} "
                 "albums, EPs, and singles for artist with ID "
                 f"{self.metadata.id},  '{self.name}'"
             )
         else:
-            self.set_albums(session)
+            self.set_albums(client)
             logger.info(
                 f"Starting attempt to get {self.albums.total_number_of_items} albums "
                 f"for artist with ID {self.metadata.id}, '{self.name}'"
@@ -102,7 +102,7 @@ def get_albums(
         for i, a in enumerate(self.albums.items):
             album: Album = Album(album_id=a.id)
             album.get(
-                session=session,
+                client=client,
                 audio_format=audio_format,
                 out_dir=out_dir,
                 metadata=a,
@@ -111,28 +111,28 @@ def get_albums(
 
     def get_videos(
         self,
-        session: Session,
+        client: Client,
         out_dir: Path,
     ) -> List[Optional[str]]:
         """This method sets self.videos by calling self.set_videos()
         then, for each video, instantiates a Video object and executes
         video.get()"""
-        self.set_videos(session)
+        self.set_videos(client)
         logger.info(
             f"Starting attempt to get {self.videos.total_number_of_items} videos "
             f"for artist with ID {self.metadata.id}, '{self.name}'"
         )
         for i, v in enumerate(self.videos.items):
             video: Video = Video(video_id=v.id)
             video.get(
-                session=session,
+                client=client,
                 out_dir=out_dir,
                 metadata=v,
             )
 
     def get(
         self,
-        session: Session,
+        client: Client,
         audio_format: AudioFormat,
         out_dir: Path,
         include_eps_singles: bool,
@@ -146,27 +146,27 @@ def get(
             4. get_albums()
         Then, if no_extra_files is False, save_artist_image()
         """
-        self.set_metadata(session)
+        self.set_metadata(client)
         if self.metadata is None:
             return
 
         self.set_artist_dir(out_dir)
-        self.get_videos(session, out_dir)
+        self.get_videos(client, out_dir)
         if include_eps_singles:
             self.get_albums(
-                session,
+                client,
                 audio_format,
                 out_dir,
                 include_eps_singles=True,
                 no_extra_files=no_extra_files,
             )
         self.get_albums(
-            session,
+            client,
             audio_format,
             out_dir,
             include_eps_singles=False,
             no_extra_files=no_extra_files,
         )
 
         if not no_extra_files:
-            self.save_artist_image(session)
+            self.save_artist_image(client)

tidal_wave/dash.py

@@ -6,7 +6,7 @@
 from xml.etree import ElementTree as ET
 
 import dataclass_wizard
-from requests import Session
+from httpx import Client
 
 from .models import TracksEndpointStreamResponseJSON
 from .utils import decrypt_manifest_key_id
@@ -86,7 +86,7 @@ def __post_init__(self):
             int(self.start_number) if self.start_number is not None else None
         )
 
-    def build_urls(self, session: Session) -> Optional[List[str]]:
+    def build_urls(self, client: Client) -> Optional[List[str]]:
         """Parse the MPEG-DASH manifest into a list of URLs. In
         particular, look for a special value, r, in self.segment_timeline.s.
         If there is no such value, set r=1. In both cases, start substituting
@@ -108,7 +108,7 @@ def sub_number(n: int, p: str = r"\$Number\$", s: str = self.media) -> str:
         if r is None:
             urls_list: List[str] = [self.initialization]
             number: int = 1
-            while session.head(url=sub_number(number)).status_code != 500:
+            while client.head(url=sub_number(number)).status_code != 500:
                 urls_list.append(sub_number(number))
                 number += 1
             else:
@@ -119,7 +119,7 @@ def sub_number(n: int, p: str = r"\$Number\$", s: str = self.media) -> str:
                 sub_number(i) for i in number_range
             ]
             number: int = r + 1
-            while session.head(url=sub_number(number)).status_code != 500:
+            while client.head(url=sub_number(number)).status_code != 500:
                 urls_list.append(sub_number(number))
                 number += 1
             else: