From 7b365e014d4941a2133a6690912c7645a030419a Mon Sep 17 00:00:00 2001 From: Romain Dorgueil Date: Mon, 16 Jul 2018 11:56:46 +0200 Subject: [PATCH] aggregates: Adds first version of Reduce() based on @levic work. --- bonobo/nodes/aggregation.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) create mode 100644 bonobo/nodes/aggregation.py diff --git a/bonobo/nodes/aggregation.py b/bonobo/nodes/aggregation.py new file mode 100644 index 0000000..f09cc17 --- /dev/null +++ b/bonobo/nodes/aggregation.py @@ -0,0 +1,16 @@ +from bonobo.config import Configurable, Method, Option, ContextProcessor, use_raw_input +from bonobo.util import ValueHolder + + +class Reduce(Configurable): + function = Method() + initializer = Option(required=False) + + @ContextProcessor + def buffer(self, context): + values = yield ValueHolder(self.initializer() if callable(self.initializer) else self.initializer) + context.send(values.get()) + + @use_raw_input + def __call__(self, values, bag): + values.set(self.function(values.get(), bag))