I wanted a type-safe pipe utility to help me write Python in a more functional style, but unfortunately toolz.pipe and returns.pipelines.flow both output Any.
Happily, it turns out creating your own pipe mechanism is a one-liner:
from functools import reduce
result = reduce(lambda acc, f: f(acc), (fn1, fn2, fn3), value)Which you can reuse by wrapping it in a function:
from functools import reduce
from typing import Callable, TypeVar
_A = TypeVar("A")
_B = TypeVar("B")
def pipe(value: _A, *functions: Callable[[_A], _A]) -> _A:
"""Pass a value through a series of functions that expect one argument of the same type."""
return reduce(lambda acc, f: f(acc), functions, value)And calling it with any number of functions:
assert pipe("1", int, float, str) == "1.0"
# => i.e. str(float(int('1')))
# => i.e. int("1") -> float(1) -> str(1.0) -> "1.0"So you can stop thinking up names for throwaway variables like these:
def str_to_float_str(value: string):
as_integer = int(value)
as_float = float(as_integer)
as_string = str(as_float)
return as_string
assert str_to_float_str("1") == "1.0"Credit to Statically-Typed Functional Lists in Python with Mypy by James Earl Douglas and the returns.pipeline.flow source code for the inspiration.
Update on Nov 29, 2024
While the pipe function above with the Callable[[A], A] type hint works fine if every function in the pipeline outputs the same type (A), the example I showed above doesn’t actually work out very well! Mypy notices that some of the functions (int and float) output a different type than we started with (str), so we aren’t actually passing A all the way through.
After trying a number of workarounds (and getting some good advice on Reddit), I learned that you can either tell Mypy what’s going on by painstakingly articulating every possible overload:
_A = TypeVar("A")
_B = TypeVar("B")
_C = TypeVar("C")
_D = TypeVar("D")
_E = TypeVar("E")
@overload
def pipe(value: _A) -> _A: ...
@overload
def pipe(value: _A, f1: Callable[[_A], _B]) -> _B: ...
@overload
def pipe(
value: _A,
f1: Callable[[_A], _B],
f2: Callable[[_B], _C]
) -> _C: ...
@overload
def pipe(
value: _A,
f1: Callable[[_A], _B],
f2: Callable[[_B], _C],
f3: Callable[[_C], _D]
) -> _D: ...
@overload
def pipe(
value: _A,
f1: Callable[[_A], _B],
f2: Callable[[_B], _C],
f3: Callable[[_C], _D],
f4: Callable[[_D], _E],
) -> _E: ...
def pipe(value: Any, *functions: Callable[[Any], Any]) -> Any:
return reduce(lambda acc, f: f(acc), functions, value)Or, you can just use expression, which already does this for you.
I’m going to do the latter, but this was a fun exercise in the meantime. 😎