Workers: Use get_*_count to get the actual number of items.

Author: tehkillerbee

tidalapi/workers.py

@@ -6,59 +6,45 @@
 
 
 def func_wrapper(args):
-    (f, offset, *args) = args
+    """
+    args: tuple(func, limit, offset, *extra_args)
+    Returns list of (index, item)
+    """
+    func, limit, offset, *extra_args = args
     try:
-        items = f(*args)
+        items = func(limit, offset, *extra_args)
     except Exception as e:
-        log.error("Failed to run %s(offset=%d, args=%s)", f, offset, args)
+        log.error(
+            "Failed to run %s(limit=%d, offset=%d, args=%s)",
+            func,
+            limit,
+            offset,
+            extra_args,
+        )
         log.exception(e)
         items = []
-    return list((i + offset, item) for i, item in enumerate(items))
+
+    return [(i + offset, item) for i, item in enumerate(items)]
 
 
 def get_items(
     func: Callable,
+    total_count: int,
     *args,
     parse: Callable = lambda _: _,
     chunk_size: int = 50,
     processes: int = 2,
 ):
     """This function performs pagination on a function that supports `limit`/`offset`
     parameters and it runs API requests in parallel to speed things up."""
+    offsets = list(range(0, total_count, chunk_size))
     items = []
-    offsets = [-chunk_size]
-    remaining = chunk_size * processes
-
-    with ThreadPoolExecutor(
-        processes, thread_name_prefix=f"mopidy-tidal-{func.__name__}-"
-    ) as pool:
-        while remaining == chunk_size * processes:
-            offsets = [offsets[-1] + chunk_size * (i + 1) for i in range(processes)]
-
-            pool_results = pool.map(
-                func_wrapper,
-                [
-                    (
-                        func,
-                        offset,
-                        chunk_size,  # limit
-                        offset,  # offset
-                        *args,  # extra args (e.g. order, order_direction)
-                    )
-                    for offset in offsets
-                ],
-            )
-
-            new_items = []
-            for results in pool_results:
-                new_items.extend(results)
 
-            remaining = len(new_items)
-            items.extend(new_items)
+    with ThreadPoolExecutor(processes) as pool:
+        args_list = [(func, chunk_size, offset, *args) for offset in offsets]
 
-    items = [_ for _ in items if _]
-    sorted_items = list(
-        map(lambda item: item[1], sorted(items, key=lambda item: item[0]))
-    )
+        for page_items in pool.map(func_wrapper, args_list):
+            items.extend(page_items)
 
-    return list(map(parse, sorted_items))
+    items = [item for _, item in sorted(items, key=lambda x: x[0])]
+    return list(map(parse, items))