Repository: tarohi24/typedflow

Issue:
Batch doesn't flow if any items in it has an error
even if the majority isn't troublesome

Discussion:

Patch:
diff --git a/typedflow/typedflow.py b/typedflow/typedflow.py
index 8185bd6..b6dc426 100644
--- a/typedflow/typedflow.py
+++ b/typedflow/typedflow.py
@@ -19,6 +19,10 @@ T = TypeVar('T')  # serializable
 K = TypeVar('K')  # serializable
 
 
+class BatchIsEmpty(Exception):
+    pass
+
+
 @dataclass
 class Batch(Generic[T]):
     batch_id: int
@@ -31,8 +35,18 @@ class Task(Generic[T, K]):
 
     def process(self,
                 batch: Batch[T]) -> Batch[K]:
-        lst: List[K] = [self.func(item) for item in batch.data]
-        return Batch(batch_id=batch.batch_id, data=lst)
+        products: List[K] = []
+        for item in batch.data:
+            try:
+                products.append(self.func(item))
+            except Exception as e:
+                logger.warn(repr(e))
+                continue
+        if len(products) > 0:
+            return Batch[K](batch_id=batch.batch_id,
+                            data=products)
+        else:
+            raise BatchIsEmpty()
 
 
 @dataclass
@@ -97,7 +111,6 @@ class Pipeline:
         for batch in self.loader.load():
             try:
                 product: Batch = _run(batch, self.pipeline)
-            except Exception as e:
-                logger.warn(repr(e))
+            except BatchIsEmpty:
                 continue
             self.dumper.dump(product)


Test patch:
diff --git a/typedflow/tests/typedflow/test_task.py b/typedflow/tests/typedflow/test_task.py
index 3294062..93e6463 100644
--- a/typedflow/tests/typedflow/test_task.py
+++ b/typedflow/tests/typedflow/test_task.py
@@ -77,7 +77,7 @@ def test_process(pl, capsys):
 def test_except_batch(invalid_pl, capsys):
     invalid_pl.run()
     out, _ = capsys.readouterr()
-    assert out == ''
+    assert out == '15\n12\n'
 
 
 def test_multibatch_process(mutlibatch_pl, capsys):