I've talked a couple of times [1] and [2] about how beautiful it's having a pipe operator in a language, though it's not particular common, and ways to simulate it in Python. Having a pipe operator makes applying functions to a value as convenient as chaining methods. When chaining methods we can leverage (if available) the safe navigation/optional chaining/elvis (?.) operator, to deal with null values. So, I've been thinking about null safety and pipes (not applying a function if the value is null, and coalescing to a default value).
In my previous post I mentioned that JavaScript had 2 different proposals for a pipe operator, but one of them has been discarded. I've been checking if this proposal includes null safety and the answer is not. It was discussed in the early stages, apart from the normal |> operator, having an additional ?|> operator for null safe cases, but it was discarded
// not null-safe, active proposal
user
|> getProfile(%)
|> formatProfile(%)
// null-safe, has been discarded
value ?|> fn
value |> fn ?? default
It was rejected on the basis that Pipelines should be pure syntax for data flow, not control flow.
To my surprise (I was not aware php continues to be used and evolve) PHP has recently added a pipe operator to the language, and for the moment it also lacks a null-safe version.
For Python decision makers adding a pipe operator seems "making the language too complex for beginners"... (you can't imagine how much I hate that so common kind of "pythonic" reflections...), but as I explained in my previous post we can easily add a pipe function that makes the trick (what has also been requested multiple times is adding such kind of function to functools, but no luck so far). An implementation is so simple as this:
def pipe(val: Any, *fns: Callable[[Any], Any]) -> Any:
"""
pipes function calls over an initial value
"""
def _call(val, fn):
return fn(val)
return functools.reduce(_call, fns, val)
And we can use it like this:
@dataclass
class Post:
id: str
title: str
author: str
def get_post(post_id: str) -> Post | None:
# simulate a function that may return None
if post_id == "1":
return Post(id="1", title="First post", author="1")
else:
return None
def get_address(person_id: str) -> str | None:
# simulate a function that may return None
if person_id == "1":
return "Rue de La Nation, Paris"
else:
return None
pipe("1",
get_post,
lambda post: get_address(post.author),
str.upper,
print,
)
# RUE DE LA NATION, PARIS
Creating a null aware equivalent is quite simple. The idea I came up with is having pipe accept not just a sequence of callables, but a sequence of callables or flag and callable or flag and value, with the flag indicating the we have to check for null before applying the Callable, or that we have to coalesce it to a value. Let's see the code:
// sentinel values
NULL_SAFE = object()
COALESCE = object()
def pipe(val: Any, *steps: Callable[[Any], Any] | tuple[Any, Callable[[Any], Any] | Any]) -> Any:
"""
pipes function calls over an initial value, with support for null safety and coalescing:
"""
def _call(val, step: Callable[[Any], Any] | tuple[Any, Callable[[Any], Any] | Any]) -> Any:
if callable(step):
return step(val)
else:
option = step[0]
if option is NULL_SAFE:
fn = step[1]
return None if val is None else fn(val)
elif option is COALESCE:
default_val = step[1]
return default_val if val is None else val
else:
raise ValueError(f"Invalid option: {option}")
return functools.reduce(_call, steps, val)
pipe2("2",
(NULL_SAFE, get_post),
(NULL_SAFE, lambda post: get_address(post.author)),
(COALESCE, "Not found"),
str.upper,
print,
)
# NOT FOUND
The function is quite minimal. We should add it proper error handing, throwing meaningful exceptions for each potential incorrect usage. You can just ask a GPT to add it and you'll end up with something like this:
def pipe(val: Any, *steps: Union[Callable[[Any], Any], Tuple[object, Any]]) -> Any:
"""
Pipe value through callables or option-tuples.
Steps can be:
- a callable: called as fn(acc)
- null_safe(fn): tuple (NULL_SAFE, fn) — only call fn if acc is not None
- coalesce(default): tuple (COALESCE, default) — replace None with default
Raises TypeError or ValueError for invalid steps.
"""
def _call(val: Any, step: Union[Callable[[Any], Any], Tuple[object, Any]]) -> Any:
if callable(step):
return step(val)
if not (isinstance(step, tuple) and len(step) == 2):
raise TypeError("pipe2 steps must be callables or 2-tuples from null_safe/coalesce")
option, payload = step
if option is NULL_SAFE:
if val is None:
return None
if not callable(payload):
raise TypeError("NULL_SAFE payload must be callable")
return payload(val)
if option is COALESCE:
default = payload
return default if val is None else val
raise ValueError(f"Unknown pipe2 option: {option!r}")
return functools.reduce(_call, steps, val)
