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)]
foldl
A alias for functools.reduce with the order of arguments fixed to be more inline with function programming standards.
New in 0.4.0
>>> from functional_pipeline import pipeline, foldl
>>> from operator import add
>>> pipeline(10, [range, (foldl, add, 0)])
45
>>> pipeline([], [(foldl, add, 0)])
0
>>> pipeline([], [iter, (foldl, add, 0)])
0
>>> pipeline(3, [range, (foldl, max, 5)])
5
scanl
An alias for itertools.accumulate with the order of the arguments fixed to be more inline with functional programming standards.
New in 0.5.0
>>> from functional_pipeline import pipeline, scanl
>>> from operator import add
>>> pipeline(5, [range, (scanl, add, 0), list])
[0, 0, 1, 3, 6, 10]
>>> pipeline(4, [range, (scanl, max, 5), list])
[5, 5, 5, 5, 5]
>>> pipeline([1, 2, 3], [(scanl, lambda x, y: 2*x + y, 4), list])
[4, 9, 20, 43]
filter_where
This is a shortcut function that allows for filter and lens to be combined with a comparison operator, so
filter_where('data.name', 'John) is functionally equivalent to (filter , lambda x: lens('data.name')(x) == 'John')
New in 0.6.0
It works on lists, dictionaries, and objects since it uses lens() as the operator for the first. It also has
a default comparator argument that is defaulted to operator.eq; This allows for filtering where Greater or Less Than.
>>> from functional_pipeline import pipeline, filter_where
>>> dataset = [
... {'name': 'John', 'meta': {'age': 30, 'gender': 'M'}},
... {'name': 'Jane', 'meta': {'age': 43, 'gender': 'F'}},
... {'name': 'James', 'meta': {'age': 22, 'gender': 'M'}},
... {'name': 'Mary', 'meta': {'age': 14, 'gender': 'f'}},
... ]
>>> pipeline(dataset, [filter_where('meta.gender', 'M'), list])
[{'name': 'John', 'meta': {'age': 30, 'gender': 'M'}}, {'name': 'James', 'meta': {'age': 22, 'gender': 'M'}}]
>>> from operator import ge
>>> pipeline(dataset, [filter_where('meta.age', 30, ge), list])
[{'name': 'John', 'meta': {'age': 30, 'gender': 'M'}}, {'name': 'Jane', 'meta': {'age': 43, 'gender': 'F'}}]