add a fold_effect function.

Author: radix

effect/fold.py

@@ -0,0 +1,33 @@
+from effect import Constant, Effect
+
+
+_sneaky = object()
+
+def fold_effect(f, initial, effects):
+    """
+    Fold over effects.
+
+    The function will be called with the accumulator (starting with
+    ``initial``) and a result of an effect repeatedly for each effect. The
+    result of the previous call will be passed as the accumulator to the next
+    call.
+
+    :param callable f: function of ``(accumulator, element) -> accumulator``
+    :param initial: The value to be passed as the accumulator to the first
+        invocation of ``f``.
+    :param effects: sequence of Effects.
+    """
+
+    def folder(acc, element):
+        # A bit of a hack here. We could just wrap `initial` in a
+        # Effect(Constant()), but to simplify testing we avoid the additional
+        # Effect by "sneaking" it in this way.
+
+        # This way people can use SequenceDispatcher and not get anything
+        # unexpected.
+        if acc is _sneaky:
+            return element.on(lambda r: f(initial, r))
+        else:
+            return acc.on(lambda r: element.on(lambda r2: f(r, r2)))
+
+    return reduce(folder, effects, _sneaky)

effect/test_fold.py

@@ -0,0 +1,21 @@
+
+import operator
+
+from effect import Constant, Effect, sync_perform
+from effect.fold import fold_effect
+from effect.testing import SequenceDispatcher
+
+
+def test_fold_effect():
+    effs = [Effect('a'), Effect('b'), Effect('c')]
+
+    dispatcher = SequenceDispatcher([
+        ('a', lambda i: 'Ei'),
+        ('b', lambda i: 'Bee'),
+        ('c', lambda i: 'Cee'),
+    ])
+    eff = fold_effect(operator.add, 'Nil', effs)
+
+    with dispatcher.consume():
+        result = sync_perform(dispatcher, eff)
+    assert result == 'NilEiBeeCee'