Functional pipeline comes with a number of helper functions to make operating in pipelines cleaner and easier.

tap

The tap function is a safe entrypoint into a pipeline that calls a given function, and returns the original value to the pipeline.

This is useful for debugging or for side effects. It creates a deepcopy to call the given function, with, so it is safe to inject this into a pipeline and have it not mutate the pipeline.

>>> from functional_pipeline import pipeline, tap

>>> results = pipeline(
...     range(20),
...     [
...         (filter, lambda x: x % 3 == 0),
...         (map, lambda x: x ** 2),
...         list,
...         tap(print), # will printout the list at this point in the pipeline
...         reversed,
...     ]
... )
[0, 9, 36, 81, 144, 225, 324]

>>> list(results) 
[324, 225, 144, 81, 36, 9, 0]

lens

The lens helper function is built to drill down into an object to quickly extract the information you might need.

Lens will try to dig into dictionaries, lists, tuples, and class instances. It takes a key or keypath (. separated), and returns a function that will retrieve that keypath on a given object. If that keypath does not exist, it will return None.

For more complex queries you can use it in conjunction with filter:

>>> from functional_pipeline import pipeline, lens

>>> people = [
...     { 
...         'first_name': 'John', 
...         'last_name': 'Smith', 
...         'age': 32, 
...         'employment': [{'name': 'McDonalds', 'position': 'Manager'}]
...     },
...     { 
...         'first_name': 'Jane', 
...         'last_name': 'Smith', 
...         'age': 30, 
...         'employment': [{'name': 'BurgerKing', 'position': 'Manager'}]
...     },
...     {
...         'first_name': 'Billy', 
...         'last_name': 'Bob', 
...         'age': 55, 
...         'employment': [{'name': 'Microsoft', 'position': 'Programmer'}]
...     },
... ]

>>> first_names = pipeline(
...     people, 
...     [ 
...         (map, lens('first_name')),
...         list,
...     ] 
... )

>>> first_names
['John', 'Jane', 'Billy']

>>> unique_last_names = pipeline(
...     people,
...     [
...         (map, lens('last_name')),
...         set,
...         list,
...         sorted,
...     ]
... )

>>> unique_last_names
['Bob', 'Smith']

>>> names_of_companies_current_employed = pipeline(
...     people,
...     [
...         (map, lens('employment.0.name')),
...         list,
...     ]
... )

>>> names_of_companies_current_employed
['McDonalds', 'BurgerKing', 'Microsoft']

>>> first_name_of_managers = pipeline(
...     people, 
...     [
...         (filter, lambda person: lens('employment.0.position')(person) == 'Manager'),
...         (map, lens('first_name')),
...         list,
...     ]
... )

>>> first_name_of_managers
['John', 'Jane']

sort

Helper method wrapping sorted that allows for the keyfunc= and reversed= attributes to be passed in

>>> from functional_pipeline import pipeline, sort

>>> coordinates = [
...     (2, 3),
...     (5, 7),
...     (-2, 1),
...     (3, 2),
... ]

>>> sorted_positive_cooridates = pipeline(
...     coordinates,
...     [
...         (filter, lambda x: x[0] > 0),
...         sort(lambda x: x[0]),
...         (map, lambda x: f"x={x[0]},y={x[1]}"),
...         list,
...     ]
... )

>>> sorted_positive_cooridates
['x=2,y=3', 'x=3,y=2', 'x=5,y=7']

take

Take is a list/generator agnostic way of taking the first n elements from a sequence.

>>> from functional_pipeline import pipeline, take

>>> five_from_range = pipeline(
...     range(100),
...     [
...         take(5)
...     ]
... )

>>> list(five_from_range) # until list() is called, this is still a generator. Take is lazy
[0, 1, 2, 3, 4]


>>> two_from_list = pipeline(
...     [1,2,3],
...     [
...         take(2),
...         list,
...     ]
... )

>>> list(two_from_list)
[1, 2]

clone

Clone is a helper function that takes an object and returns a tuple of two of the object.

>>> from functional_pipeline import pipeline, clone

>>> words = ['racecar', 'palindrome', 'lol', 'python', 'wow']

>>> palindromes = pipeline(
...     words,
...     [
...         (map, clone),
...         (map, lambda x: (x[0], x[1][::-1])),
...         (filter, lambda x: x[0] == x[1]),
...         (map, lambda x: x[1]),
...         list,
...     ]
... )

>>> palindromes
['racecar', 'lol', 'wow']

index

The index function is a safe get of an index of a sequence

>>> from functional_pipeline import pipeline, index

>>> success = pipeline(
...     [1, 2, 3],
...     [
...         index(1),    
...     ]    
... )

>>> success
2

>>> safe = pipeline(
...     [1, 2, 3],
...     [
...         index(20),    
...     ]    
... )

>>> safe

contains

Predicate that wraps the in keyword.

>>> from functional_pipeline import pipeline, contains

>>> words = ['this', 'that', 'other', 'foo', 'bar']

>>> words_with_t = pipeline(
...     words,
...     [
...         (filter, contains('t')),
...         list,
...     ]
... )

>>> words_with_t
['this', 'that', 'other']

join

Wrapper around str.join().

>>> from functional_pipeline import pipeline, join

>>> piped_evens = pipeline(
...     range(20),
...     [
...         (filter, lambda x: x % 2 == 0),
...         (map, str),
...         join('|'),
...     ]
... )

>>> piped_evens
'0|2|4|6|8|10|12|14|16|18'

flatten

Flatten an iterable of iterables into a single iterable.

Will return a generator of the flattened state.

>>> from functional_pipeline import pipeline, flatten

>>> nested_ranges = (range(n) for n in range(5))

>>> flat = pipeline(
...     nested_ranges,
...     [
...         flatten,
...         list,
...     ]
... )

>>> flat
[0, 0, 1, 0, 1, 2, 0, 1, 2, 3]

not_none

Shorthand for a predicate that checks if obj is not None

>>> from functional_pipeline import pipeline, not_none

>>> stream = [1, None, 2, 3, 4, None, 5, 6]

>>> output = pipeline(
...     stream,
...     [
...         (filter, not_none),
...         list,
...     ]
... )

>>> output
[1, 2, 3, 4, 5, 6]

flatmap

A shorthand for map -> flatten.

New in 0.3.0

>>> from functional_pipeline import pipeline, flatmap

>>> results = pipeline(
...     [3, 4, 2],
...     [
...         (flatmap, range),
...         list,
...     ]
... )

>>> results
[0, 1, 2, 0, 1, 2, 3, 0, 1]

zipmap

Takes a tuple of functions, and applies a list of tuples to them, returning the tuple outputs.

New in 0.3.0

>>> from functional_pipeline import pipeline, zipmap

>>> def double(x):
...     return x * 2

>>> def square(x):
...     return x ** 2

>>> results = pipeline(
...     5,
...     [
...         range,
...         enumerate,
...         (zipmap, (double, square)),
...         list,
...     ]
... )

>>> results
[(0, 0), (2, 1), (4, 4), (6, 9), (8, 16)]