Skip to content
Open
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions DIRECTORY.md
Original file line number Diff line number Diff line change
Expand Up @@ -856,6 +856,7 @@
* [Nested Brackets](other/nested_brackets.py)
* [Number Container System](other/number_container_system.py)
* [Password](other/password.py)
* [Pipeline](other/pipeline.py)
* [Quine](other/quine.py)
* [Scoring Algorithm](other/scoring_algorithm.py)
* [Sdes](other/sdes.py)
Expand Down
46 changes: 46 additions & 0 deletions other/pipeline.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
from collections.abc import Callable, Sequence
from typing import Any, TypeVar

T = TypeVar("T")


class Pipeline:
"""
Functional Programming implementation of a Pipeline with Unix Pipe Syntax.
Instead of using the "dot" notation for applying a function on a given object,
it uses `|` inspired from unix pipeline.

Examples:
>>> pipeline = Pipeline()
>>> pipeline = pipeline | (lambda x: x + 1) | (lambda x: x * 2)
>>> pipeline(1)
4
>>> pipeline = Pipeline() | (lambda x: x * x) | (lambda x: x - 3)
>>> pipeline(3)
6
>>> from functools import reduce
>>> def f1(ls): return map(lambda x: x**2, ls)
>>> def f2(ls): return filter(lambda x: x % 2 == 0, ls)
>>> def f3(ls): return reduce(lambda x, y: x + y, ls)
>>> pipeline = Pipeline() | f1 | f2 | f3
>>> pipeline([1, 2, 3, 4])
20
"""

def __init__(self, f_ls: Sequence[Callable] | None = None):
self._f_ls = f_ls or []

def __or__(self, other: Callable) -> "Pipeline":
return Pipeline(f_ls=[*self._f_ls, other])

def __call__(self, x: T, f_ls_: Sequence[Callable] | None = None) -> Any:
f_ls = f_ls_ or self._f_ls
if len(f_ls) == 1:
return f_ls[0](x)
return self(f_ls[0](x), f_ls_=f_ls[1:])


if __name__ == "__main__":
import doctest

doctest.testmod()
Loading