Removed dependency on aioreactive. Doing this my way.

Signed-off-by: Cliff Hill <xlorep@darkhelm.org>

src/playlist/const.py

@@ -9,6 +9,8 @@ __all__ = ["APPNAME", "APPAUTHOR", "PATHS"]
 APPNAME = "plex-playlist"
 APPAUTHOR = "Cliff Hill"
 
+DOWNLOAD_BATCH_SIZE = 1000
+
 
 @dataclasses.dataclass
 class Paths:

src/playlist/models.py

@@ -0,0 +1,70 @@
+"""Contains the data models used throughout the application."""
+import dataclasses
+import datetime
+import functools
+import typing
+
+import desert  # type: ignore [import]
+import marshmallow  # type: ignore [import]
+import plexapi.audio  # type: ignore [import]
+
+
+class SchemaMeta(type):
+    """Metaclass to access the schema for the given class."""
+
+    schema: typing.ClassVar[marshmallow.schema.Schema]
+
+    @property  # type: ignore [no-redef,misc]
+    @functools.cache
+    def schema(cls) -> marshmallow.schema.Schema:
+        """Return the schema object for this class."""
+        return desert.schema(cls)
+
+
+@dataclasses.dataclass(frozen=True)
+class Track(metaclass=SchemaMeta):
+    """Model defining a Track object."""
+
+    id: int
+    track_num: int
+    title: str
+    artist: str
+    album_num: int
+    album: str
+    album_artist: str
+    duration: int
+    rating: typing.Optional[int]
+    comments: str
+    added: datetime.datetime
+    play_count: int
+    played: typing.Optional[datetime.datetime]
+
+    @classmethod
+    def load(
+        cls: typing.Type["Track"],
+        data: dict[str, typing.Union[int, str, None]],
+    ) -> "Track":
+        """Load the given data into a Track instance."""
+        return typing.cast(Track, cls.schema.load(data))
+
+    @classmethod
+    def from_plex(
+        cls: typing.Type["Track"],
+        track: plexapi.audio.Track,
+    ) -> "Track":
+        """Convert a Plex track object to an internal Track object."""
+        return cls(
+            id=track.ratingKey,
+            track_num=track.index,
+            title=track.title,
+            artist=track.artist().title,
+            album_num=track.parentIndex,
+            album=track.parentTitle,
+            album_artist=track.grandparentTitle,
+            duration=track.duration,
+            rating=track.userRating,
+            comments=track.summary,
+            added=track.addedAt,
+            play_count=track.viewCount,
+            played=track.lastViewedAt,
+        )

src/playlist/plex.py

@@ -1,79 +1,69 @@
 """Contains the code to communicate to the Plex server."""
+import asyncio
+import functools
 import typing
 
-import aioreactive as rx  # type: ignore
-import plexapi.audio  # type: ignore
-import plexapi.server  # type: ignore
+import plexapi.audio  # type: ignore [import]
+import plexapi.server  # type: ignore [import]
 
+from playlist import const
+from playlist import models
 from playlist import settings
 
-__all__ = ["server", "gen_tracks"]
+__all__ = ["server", "gen_tracks", "total_track_count"]
 
 server = plexapi.server.PlexServer(**settings.get().creds.to_dict())
 
 
-def gen_batched_tracks(
+def _downloader(
+    pos: int,
+    size: int,
+    loop: asyncio.events.AbstractEventLoop,
+    search_tracks: typing.Callable,  # type: ignore [type-arg]
+) -> typing.Awaitable[list[plexapi.audio.Track]]:
+    return loop.run_in_executor(
+        None,
+        functools.partial(
+            search_tracks,
+            maxresults=size,
+            container_start=pos,
+            container_size=size,
+        ),
+    )
+
+
+async def gen_tracks(
     *,
-    batch_size: int = 100,
-) -> typing.Generator[list[plexapi.audio.Track], None, None]:
-    """Generate Tracks from the server in `batch_size` batches."""
-    pos = 0
-    while 1:
-        batch = list(
-            server.library.section("Music").searchTracks(
-                maxresults=batch_size,
-                container_start=pos,
-                container_size=batch_size,
-            ),
-        )
-        yield batch
-        if len(batch) < batch_size:
-            break
-        pos += batch_size
-
-
-def track_mapper(
-    tracks: list[plexapi.audio.Track],
-) -> list[dict[str, typing.Any]]:
-    return [
-        {
-            "id": track.ratingKey,
-            "track_num": track.index,
-            "title": track.title,
-            "artist": track.artist().title,
-            "album_num": track.parentIndex,
-            "album": track.parentTitle,
-            "album_artist": track.grandparentTitle,
-            "duration": track.duration,
-            "rating": track.userRating,
-            "comments": track.summary,
-            "added": track.addedAt,
-            "play_count": track.viewCount,
-            "played": track.lastViewedAt,
-        }
-        for track in tracks
-    ]
-
-
-async def to_obs(
-    tracks: list[dict[str, typing.Any]],
-) -> rx.AsyncRx[dict[str, typing.Any]]:
-    return rx.AsyncRx.from_iterable(iter(tracks))
-
-
-def gen_tracks(
-    *,
-    batch_size: int = 100,
-) -> rx.AsyncIteratorObserver[dict[str, typing.Any]]:
+    batch_size: int = const.DOWNLOAD_BATCH_SIZE,
+) -> typing.AsyncGenerator[models.Track, None]:
     """Generate all Tracks from the Server, asynchronously.
 
     Keyword Args:
         batch_size: determines how many Tracks are pulled from the Server at a time.
     """
-    source = rx.AsyncRx.from_iterable(gen_batched_tracks(batch_size=batch_size))
-    track_rx = rx.pipe(
-        source,
-        rx.map(track_mapper),
-        rx.flat_map_latest_async(to_obs),
+    quotient, remainder = divmod(await total_track_count(), batch_size)
+    loop = asyncio.get_running_loop()
+    search_tracks = server.library.section("Music").searchTracks
+    batches = [
+        _downloader(ndx, batch_size, loop, search_tracks) for ndx in range(quotient)
+    ]
+    batches.append(_downloader(quotient, remainder, loop, search_tracks))
+    for batch in asyncio.as_completed(batches):
+        batch_gen = (models.Track.from_plex(plex_track) for plex_track in await batch)
+        for track in batch_gen:
+            yield track
+
+
+async def total_track_count() -> int:
+    """Get the total number of tracks in the server."""
+    loop = asyncio.get_running_loop()
+    num_tracks: int = await loop.run_in_executor(
+        None,
+        (
+            lambda: typing.cast(
+                int,
+                server.library.section("Music").totalViewSize("track"),
+            )
+        ),
     )
-    return rx.AsyncIteratorObserver(track_rx)
+    return num_tracks

src/playlist/utils.py

@@ -0,0 +1,15 @@
+"""Utility functions & classes used in the application."""
+import contextlib
+import datetime
+import typing
+
+
+@contextlib.contextmanager
+def time_this() -> typing.Generator[None, None, None]:
+    """Context manager to time the execution of a block of code."""
+    start = datetime.datetime.now()
+    try:
+        yield
+    finally:
+        end = datetime.datetime.now()
+        print(f"Elapsed time: {end - start}")