WIP: async experiments

plextraktsync/commands/sync.py

@@ -15,11 +15,11 @@
 from plextraktsync.walker import WalkConfig, Walker
 
 
-def sync_all(walker: Walker, plex: PlexApi, runner: Sync, dry_run: bool):
+async def sync_all(walker: Walker, plex: PlexApi, runner: Sync, dry_run: bool):
     click.echo(f"Plex Server version: {plex.version}, updated at: {plex.updated_at}")
     click.echo(f"Server has {len(plex.library_sections)} libraries: {plex.library_section_names}")
 
-    runner.sync(walker, dry_run=dry_run)
+    await runner.sync(walker, dry_run=dry_run)
 
 
 @click.command()
@@ -120,6 +120,6 @@ async def sync(
 
     with measure_time("Completed full sync"):
         try:
-            sync_all(walker=w, plex=plex, runner=factory.sync(), dry_run=dry_run)
+            await sync_all(walker=w, plex=plex, runner=factory.sync(), dry_run=dry_run)
         except RuntimeError as e:
             raise ClickException(str(e))

plextraktsync/factory.py

@@ -69,14 +69,18 @@ def sync(self):
     @memoize
     def progressbar(self, enabled=True):
         if enabled:
-            import warnings
-
-            from tqdm import TqdmExperimentalWarning
-            from tqdm.rich import tqdm
-
-            warnings.filterwarnings("ignore", category=TqdmExperimentalWarning)
-
-            return tqdm
+            from tqdm.asyncio import tqdm_asyncio
+
+            return tqdm_asyncio
+
+            # import warnings
+            #
+            # from tqdm import TqdmExperimentalWarning
+            # from tqdm.rich import tqdm
+            #
+            # warnings.filterwarnings("ignore", category=TqdmExperimentalWarning)
+            #
+            # return tqdm
 
         return None
 

plextraktsync/media.py

@@ -119,7 +119,7 @@ def __init__(self, plex: PlexApi, trakt: TraktApi):
         self.plex = plex
         self.trakt = trakt
 
-    def resolve_any(self, pm: PlexLibraryItem, tm=None):
+    async def resolve_any(self, pm: PlexLibraryItem, tm=None):
         try:
             guids = pm.guids
         except (PlexApiException, RequestException) as e:

plextraktsync/plex_api.py

@@ -394,7 +394,7 @@ def find_by_id(self, id: str):
         except NotFound:
             return None
 
-    def all(self, max_items: int):
+    async def all(self, max_items: int):
         libtype = self.section.TYPE
         key = self.section._buildSearchKey(libtype=libtype, returnKwargs=False)
         start = 0
@@ -405,7 +405,8 @@ def all(self, max_items: int):
             if not len(items):
                 break
 
-            yield from items
+            for m in items:
+                yield m
 
             start += size
             if start > max_items:
@@ -415,8 +416,8 @@ def all(self, max_items: int):
     def fetch_items(self, key: str, size: int, start: int):
         return self.section.fetchItems(key, container_start=start, container_size=size)
 
-    def items(self, max_items: int):
-        for item in self.all(max_items):
+    async def items(self, max_items: int):
+        async for item in self.all(max_items):
             yield PlexLibraryItem(item)
 
     def __repr__(self):

plextraktsync/sync.py

@@ -57,7 +57,7 @@ class Sync:
     def __init__(self, config: Config):
         self.config = SyncConfig(config)
 
-    def sync(self, walker: Walker, dry_run=False):
+    async def sync(self, walker: Walker, dry_run=False):
         listutil = TraktListUtil()
         trakt = walker.trakt
 
@@ -68,17 +68,22 @@ def sync(self, walker: Walker, dry_run=False):
             for lst in trakt.liked_lists:
                 listutil.addList(lst['username'], lst['listname'])
 
-        for movie in walker.find_movies():
-            self.sync_collection(movie, dry_run=dry_run)
-            self.sync_ratings(movie, dry_run=dry_run)
-            self.sync_watched(movie, dry_run=dry_run)
-            listutil.addPlexItemToLists(movie)
-
-        for episode in walker.find_episodes():
-            self.sync_collection(episode, dry_run=dry_run)
-            self.sync_ratings(episode, dry_run=dry_run)
-            self.sync_watched(episode, dry_run=dry_run)
-            listutil.addPlexItemToLists(episode)
+        i = 1
+        async for movie in walker.find_movies():
+            print(movie)
+            if i >= 4:
+                break
+            i += 1
+        #     self.sync_collection(movie, dry_run=dry_run)
+        #     self.sync_ratings(movie, dry_run=dry_run)
+        #     self.sync_watched(movie, dry_run=dry_run)
+        #     listutil.addPlexItemToLists(movie)
+        #
+        # async for episode in walker.find_episodes():
+        #     self.sync_collection(episode, dry_run=dry_run)
+        #     self.sync_ratings(episode, dry_run=dry_run)
+        #     self.sync_watched(episode, dry_run=dry_run)
+        #     listutil.addPlexItemToLists(episode)
 
         if not dry_run:
             with measure_time("Updated plex watchlist"):

plextraktsync/walker.py

@@ -180,7 +180,7 @@ def print_plan(self, print=print):
         if self.plan.shows:
             print(f"Sync Shows: {self.plan.shows}")
 
-    def get_plex_movies(self):
+    async def get_plex_movies(self):
         """
         Iterate over movie sections unless specific movie is requested
         """
@@ -189,42 +189,47 @@ def get_plex_movies(self):
         else:
             movies = self.media_from_sections(self.plan.movie_sections)
 
-        yield from movies
+        async for m in movies:
+            yield m
 
-    def find_movies(self):
-        for plex in self.get_plex_movies():
-            movie = self.mf.resolve_any(plex)
+    async def find_movies(self):
+        async for plex in self.get_plex_movies():
+            movie = await self.mf.resolve_any(plex)
             if not movie:
                 continue
             yield movie
 
-    def get_plex_shows(self):
+    async def get_plex_shows(self):
         if self.plan.shows:
             shows = self.media_from_items("show", self.plan.shows)
         else:
-            shows = self.media_from_sections(self.plex.show_sections())
+            shows = await self.media_from_sections(self.plex.show_sections())
 
-        yield from shows
+        for m in shows:
+            yield m
 
-    def find_episodes(self):
-        for plex in self.get_plex_shows():
+    async def find_episodes(self):
+        for plex in await self.get_plex_shows():
             show = self.mf.resolve_any(plex)
             if not show:
                 continue
-            yield from self.episode_from_show(show)
+            for m in self.episode_from_show(show):
+                yield m
 
-    def media_from_sections(self, sections: List[PlexLibrarySection], titles: List[str] = None):
+    async def media_from_sections(self, sections: List[PlexLibrarySection], titles: List[str] = None):
         if titles:
             # Filter by matching section names
             sections = [x for x in sections if x.title in titles]
 
         for section in sections:
             with measure_time(f"{section.title} processed"):
                 total = len(section)
-                it = self.progressbar(section.items(total), total=total, desc=f"Processing {section.title}")
-                yield from it
+                items = section.items(total)
+                it = self.progressbar(items, total=total, desc=f"Processing {section.title}")
+                async for m in it:
+                    yield m
 
-    def media_from_items(self, libtype: str, items: List):
+    async def media_from_items(self, libtype: str, items: List):
         it = self.progressbar(items, desc=f"Processing {libtype}s")
         for m in it:
             yield PlexLibraryItem(m)
@@ -245,10 +250,12 @@ def episode_from_show(self, show: Media):
             me.show = show
             yield me
 
-    def progressbar(self, iterable, **kwargs):
+    async def progressbar(self, iterable, **kwargs):
         if self._progressbar:
             pb = self._progressbar(iterable, **kwargs)
             with pb as it:
-                yield from it
+                async for m in it:
+                    yield m
         else:
-            yield from iterable
+            async for m in iterable:
+                yield m

tests/test_async.py

@@ -1,4 +1,6 @@
 #!/usr/bin/env python3 -m pytest
+import asyncio
+
 import pytest
 
 from plextraktsync.sync import Sync
@@ -8,8 +10,24 @@
 plex = factory.plex_api()
 
 
+async def items(max_items: int):
+    for item in range(max_items):
+        yield item
+
+
+async def test_async_generator():
+    async for item in items(3):
+        print(item)
+    # print(c)
+
+
 @pytest.mark.asyncio
 async def test_plex_search():
     walker: Walker = factory.walker()
     runner: Sync = factory.sync()
     await runner.sync(walker, dry_run=True)
+
+
+if __name__ == '__main__':
+    # asyncio.run(test_async_generator())
+    asyncio.run(test_plex_search())