Generic Node Functions

The canal.nodes module provides some generic, reusable node functions that can be composed to solve higher-level problems.

canal.nodes.aggregate(key_func, close_func)

Evaluate key_func(value) for each value and then accumulate a list of all values with a matching key until close_func(list) returns a true value, at which point the entire list will be emitted.

It is the caller’s responsibility to ensure that all sequences are closed. Unclosed sequences will remain in memory until the containing graph is terminated, at which point they will be discarded.

canal.nodes.grep(cond)

Emit only values for which cond(value) returns a true value, dropping all other values.

canal.nodes.json_decode(**kwargs)

Pass all values (which should be strings) to json.loads() using the provided keyword arguments and emit the resulting values.

canal.nodes.json_encode(**kwargs)

Pass all values to json.dumps() using the provided keyword arguments and emit the resulting strings.

canal.nodes.sink_to_callable(func)

Pass each value as a single positional argument into the provided callable func. The return value, if any, is discarded.

canal.nodes.transform(func)

Evaluate func(value) for each value and emit the result.

canal.nodes.uniq(cmp_func=None)

Drop any values that are equal to the previous value. If the input is a sorted list then this will remove all duplicates.

If cmp_func is provided then it is called for each value to obtain the value that will be used for comparison. This allows comparison of only part of a value (for example, an id property) rather than of the whole value.