Merge pull request #44 from radix/parallel-all-errors-alternative

Add a `parallel_all_errors` function which provides all errors instead of just FirstError

Author: radix

effect/_intents.py

@@ -55,7 +55,7 @@ def parallel(effects):
     their results, in the same order as the input to this function. If any
     child effect fails, the first such failure will be propagated as a
     :obj:`FirstError` exception. If additional error information is desired,
-    error handler callbacks can be attached to each child effect.
+    use :func:`parallel_all_errors`.
 
     :param effects: Effects which should be performed in parallel.
     :return: An Effect that results in a list of results, or which fails with
@@ -64,6 +64,25 @@ def parallel(effects):
     return Effect(ParallelEffects(list(effects)))
 
 
+def parallel_all_errors(effects):
+    """
+    Given multiple Effects, return one Effect that represents the aggregate of
+    all of their effects.  The result of the aggregate Effect will be a list of
+    their results, in the same order as the input to this function.
+
+    :param effects: Effects which should be performed in parallel.
+    :return: An Effect that results in a list of ``(is_error, result)`` tuples,
+        where ``is_error`` is True if the child effect raised an exception, in
+        which case ``result`` will be an exc_info tuple. If ``is_error`` is
+        False, then ``result`` will just be the result as provided by the child
+        effect.
+    """
+    effects = [effect.on(success=lambda r: (False, r),
+                         error=lambda e: (True, e))
+               for effect in effects]
+    return Effect(ParallelEffects(list(effects)))
+
+
 @attributes(['exc_info', 'index'])
 class FirstError(Exception):
     """

effect/test_intents.py

@@ -1,16 +1,25 @@
 from __future__ import print_function, absolute_import
 
+from functools import partial
+
+import six
+
 from testtools import TestCase
+from testtools.matchers import Equals, MatchesListwise
 
 from ._base import Effect
-from ._sync import sync_perform
-from ._dispatcher import TypeDispatcher
-
+from ._dispatcher import ComposedDispatcher, TypeDispatcher
 from ._intents import (
+    base_dispatcher,
     Constant, perform_constant,
     Error, perform_error,
     Func, perform_func,
-    FirstError)
+    FirstError,
+    ParallelEffects, parallel_all_errors)
+from ._sync import sync_perform
+from ._test_utils import MatchesReraisedExcInfo, get_exc_info
+from .async import perform_parallel_async
+from .test_parallel_performers import EquitableException
 
 
 class IntentTests(TestCase):
@@ -58,3 +67,36 @@ def test_first_error_str(self):
         self.assertEqual(
             str(fe),
             '(index=150) ValueError: foo')
+
+
+class ParallelAllErrorsTests(TestCase):
+    """Tests for :func:`parallel_all_errors`."""
+
+    def test_parallel_all_errors(self):
+        """
+        Exceptions raised from child effects get turned into (True, exc_info)
+        results.
+        """
+        exc_info1 = get_exc_info(EquitableException(message='foo'))
+        reraise1 = partial(six.reraise, *exc_info1)
+        exc_info2 = get_exc_info(EquitableException(message='bar'))
+        reraise2 = partial(six.reraise, *exc_info2)
+
+        dispatcher = ComposedDispatcher([
+            TypeDispatcher({
+                ParallelEffects: perform_parallel_async,
+            }),
+            base_dispatcher])
+        es = [Effect(Func(reraise1)),
+              Effect(Constant(1)),
+              Effect(Func(reraise2))]
+        eff = parallel_all_errors(es)
+        self.assertThat(
+            sync_perform(dispatcher, eff),
+            MatchesListwise([
+                MatchesListwise([Equals(True),
+                                 MatchesReraisedExcInfo(exc_info1)]),
+                Equals((False, 1)),
+                MatchesListwise([Equals(True),
+                                 MatchesReraisedExcInfo(exc_info2)]),
+            ]))