From 852038b3bfdfac40d19317cbd0d5c3b8773c00d7 Mon Sep 17 00:00:00 2001 From: CedricAnover Date: Wed, 4 Dec 2024 12:11:16 +0800 Subject: [PATCH 1/3] feat(other): add functional programming pipeline implementation with unix pipeline syntax --- other/pipeline.py | 46 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) create mode 100644 other/pipeline.py diff --git a/other/pipeline.py b/other/pipeline.py new file mode 100644 index 000000000000..0e3fa78a4ea9 --- /dev/null +++ b/other/pipeline.py @@ -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() From 5da5a7dc871b9ce6ba2e3344be74cc2e2ce1b4fa Mon Sep 17 00:00:00 2001 From: CedricAnover Date: Wed, 4 Dec 2024 04:12:13 +0000 Subject: [PATCH 2/3] updating DIRECTORY.md --- DIRECTORY.md | 1 + 1 file changed, 1 insertion(+) diff --git a/DIRECTORY.md b/DIRECTORY.md index d234d366df06..4fd0f952aa19 100644 --- a/DIRECTORY.md +++ b/DIRECTORY.md @@ -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) From 2e2580031f07aa8e9aea628c1bfdb797d58a4a5d Mon Sep 17 00:00:00 2001 From: CedricAnover Date: Thu, 5 Dec 2024 13:39:48 +0800 Subject: [PATCH 3/3] update and clean code --- other/pipeline.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/other/pipeline.py b/other/pipeline.py index 0e3fa78a4ea9..85a7efbe518c 100644 --- a/other/pipeline.py +++ b/other/pipeline.py @@ -27,17 +27,17 @@ class Pipeline: 20 """ - def __init__(self, f_ls: Sequence[Callable] | None = None): + def __init__(self, f_ls: Sequence[Callable] | None = 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: + def __call__(self, input_object: 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:]) + return f_ls[0](input_object) + return self(f_ls[0](input_object), f_ls_=f_ls[1:]) if __name__ == "__main__":