Updating stuff.

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

src/playlist/plex.py

@@ -32,28 +32,39 @@ def gen_batched_tracks(
         pos += batch_size
 
 
-async def track_mapper(track: plexapi.audio.Track) -> 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,
-    }
+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 gen_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,
-) -> typing.AsyncGenerator[plexapi.audio.Track, None]:
+) -> rx.AsyncIteratorObserver[dict[str, typing.Any]]:
     """Generate all Tracks from the Server, asynchronously.
 
     Keyword Args:
@@ -62,9 +73,7 @@ async def gen_tracks(
     source = rx.AsyncRx.from_iterable(gen_batched_tracks(batch_size=batch_size))
     track_rx = rx.pipe(
         source,
-        rx.flat_map(rx.AsyncRx.from_iterable),
-        rx.map_async(track_mapper),
+        rx.map(track_mapper),
+        rx.flat_map_latest_async(to_obs),
     )
-    track_obs = rx.AsyncIteratorObserver(track_rx)
-    async for track in track_obs:
-        yield track
+    return rx.AsyncIteratorObserver(track_rx)