Empty batch error fixed (#23)

src/prefecto/concurrency.py

@@ -79,6 +79,7 @@ def _make_batches(self, **params) -> list[dict[str, list[Any]]]:
 
         batches = []
 
+        i = 0
         for i in range(length // self.size):
             batch = {p: [] for p in parameters}
             for p in parameters:
@@ -163,6 +164,8 @@ def _map(self, batches: list[dict[str, list[Any]]]) -> list[PrefectFuture]:
         Returns:
             A list of futures for each batch.
         """
+        if len(batches) == 0:
+            return []
         logger = logging.get_prefect_or_default_logger()
         results: list[PrefectFuture] = []
         for i, batch in enumerate(batches[:-1]):

tests/test_concurrency.py

@@ -4,6 +4,7 @@
 """
 from __future__ import annotations
 
+import pytest
 from prefect import flow, task
 
 from prefecto.concurrency import BatchTask
@@ -23,7 +24,15 @@ def test_make_batches(self):
         batches = BatchTask(add, 3)._make_batches(a=[1, 2, 3, 4, 5], b=[2, 3, 4, 5, 6])
         assert batches == [{"a": [1, 2, 3], "b": [2, 3, 4]}, {"a": [4, 5], "b": [5, 6]}]
 
-    def test_map(self, harness):
+    @pytest.mark.parametrize(
+        "a,b,expectation",
+        [
+            ([1, 2, 3, 4, 5], [2, 3, 4, 5, 6], [3, 5, 7, 9, 11]),
+            ([1, 2], [2, 3], [3, 5]),
+            ([], [], []),
+        ],
+    )
+    def test_map(self, a: list[int], b: list[int], expectation: list[int], harness):
         """Test `BatchTask.map`."""
 
         @task
@@ -34,8 +43,8 @@ def realize(futures: list[int]):
         @flow
         def test() -> list[int]:
             """Test flow."""
-            futures = BatchTask(add, 3).map([1, 2, 3, 4, 5], [2, 3, 4, 5, 6])
+            futures = BatchTask(add, 3).map(a, b)
             return realize(futures)
 
         result = test()
-        assert result == [3, 5, 7, 9, 11]
+        assert result == expectation