Pipeline: proper implements 'errors=ignore option

Author: jsbueno

extraasync/pipeline.py

@@ -119,7 +119,7 @@ async def _sync_to_async_iterable() -> t.AsyncIterable[T]:
     return _sync_to_async_iterable()
 
 
-PipelineErrors = t.Literal["strict", "ignore", "lazy_raise"]
+PipelineErrors = t.Literal["eager", "ignore", "group"]
 
 
 class RateLimiter:
@@ -244,7 +244,7 @@ def _collect_result(self, task, next_):
             return next_((task.order_tag, task.result()))
         # Hardcoded: exception - ignore.
         logger.error("Exception in pipelined stage: %s", exc)
-        self.parent.output.put_nowait((EXC_MARKER, exc))
+        self.parent.output.put_nowait((EXC_MARKER, (task.order_tag, exc)))
 
     def _create_task(self, value: tuple[int, t.Any]):
 
@@ -312,7 +312,7 @@ def __init__(
         max_concurrency: t.Optional[int] = None,
         rate_limit: None | RateLimiter | Real = None,
         rate_limit_unit: TIME_UNIT = "second",
-        on_error: PipelineErrors = "strict",
+        on_error: PipelineErrors = "eager",
         preserve_order: bool = False,
         max_simultaneous_records: t.Optional[int] = None,
         sink: None | SupportsRShift | MutableSequence | MutableSet = None,
@@ -326,7 +326,7 @@ def __init__(
                 (i.e. if there are 2 stages, and max_concurrency is set to 4, we may have
                 up to 8 concurrent tasks running at once in the pipeline, but each stage is
                 limited to 4)
-            - on_error: WHat to do if any stage raises an exeception - defaults to re-raise the
+            - on_error: WHat to do if any stage raises an exception - defaults to re-raise the
                     exception and stop the whole pipeline
             - rate_limit: An overall rate-limitting parameter which can be used to throtle all stages.
                     If anyone stage should have a limit different from the limit to the whole pipeline,
@@ -390,7 +390,7 @@ def chain_data(self, data_source):
         # TBD
 
     async def __aiter__(self):
-        """Each iteration retrieves the next final result, after passing it trhough all teh stages
+        """Each iteration retrieves the next final result, after passing it trhough all the stages
 
         NB: calling this a single time will trigger the Pipeline background execution, and
         more than one item can be (or will be) fectched  from source in a single iteration,
@@ -429,9 +429,11 @@ async def __aiter__(self):
                 if order_marker is EXC_MARKER:
                     if self.on_error == "ignore":
                         await asyncio.sleep(0)
+                        if self.preserve_order:
+                            self.ordered_results.push((result_data[0], EXC_MARKER))
                         continue
                     elif self.on_error == "strict":
-                        raise result_data
+                        raise result_data[1]
                     elif self.on_error == "lazy":
                         raise NotImplementedError("Lazy error raising in pipeline")
                 if not self.preserve_order:
@@ -440,7 +442,8 @@ async def __aiter__(self):
                     self.ordered_results.push((order_marker, result_data))
             if self.ordered_results.peek() == last_yielded_index + 1:
                 last_yielded_index, result_data = self.ordered_results.pop()
-                yield result_data
+                if result_data is not EXC_MARKER:
+                    yield result_data
 
             await asyncio.sleep(0)
 

tests/test_pipeline.py

@@ -431,6 +431,92 @@ async def map_function(n):
     assert set(results) == set(range(0, 2 * task_amount, 2))
 
 
+@pytest.mark.parametrize(
+    ["task_ammount", "failing_steps", "two_stages", "preserve_order"],
+    [
+        (
+            10,
+            {
+                5,
+            },
+            False,
+            True,
+        ),
+        (
+            10,
+            {
+                5,
+            },
+            True,
+            True,
+        ),
+        (
+            10,
+            {
+                5,
+            },
+            True,
+            False,
+        ),
+        (
+            10,
+            {
+                5,
+            },
+            False,
+            False,
+        ),
+        (10, {1, 3, 6, 9}, False, True),
+    ],
+)
+@pytest.mark.asyncio
+async def test_pipeline_runs_to_completion_when_ignoring_exceptions(
+    task_ammount, failing_steps, two_stages, preserve_order
+):
+
+    async def producer(n, interval=0):
+        for i in range(n):
+            yield i
+            await asyncio.sleep(interval)
+
+    async def map_function(n):
+        if n in failing_steps:
+            raise ValueError()
+        await asyncio.sleep(0)
+        return n * 2
+
+    async def second_stage(n):
+        await asyncio.sleep(0)
+        return n
+
+    use_second_stage = (second_stage,) if two_stages else ()
+
+    results = []
+
+    try:
+        async with asyncio.timeout(0.1):
+            async for result in Pipeline(
+                producer(10),
+                map_function,
+                *use_second_stage,
+                preserve_order=preserve_order,
+                on_error="ignore",
+            ):
+                results.append(result)
+    except TimeoutError:
+
+        assert False, f"Timed out waiting pipeline to complete. Results: {results}"
+
+    expected = [
+        item * 2 for item in range(0, task_ammount) if item not in failing_steps
+    ]
+    if not preserve_order:
+        results = set(results)
+        expected = set(expected)
+
+    assert results == expected
+
+
 @pytest.mark.skip
 @pytest.mark.asyncio
 async def test_pipeline_max_simultaneous_record_limit(): ...