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: 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. 😎