A New library for type-safe data pipelines in Python

Lan — Mon, 31 Mar 2025 20:21:26 +0000

Hello everyone! I'm writing this post partly to spark a discussion on how functional programming can be done in python. I've been interested in being explicit with my types in python and capture errors in the types rather than silently throwing an exception without this being clear in the type.

In the spirit of writing functional-style code, often we use maps, filters, folds, as well as generally a richer composition of expressions. In order to make this more feasible to write in python, I've introduced checkpipe. The gap this fills is that it allows you to express the types of your pipeline stages to python which aids makes any error clear through the language server or type checkers like mypy.

I would appreciate if you check out the repository at
https://github.com/LanHikari22/checkpipe
It has many extensive examples in the README.md. You can also install it and try it out with python3 -m pip install checkpipe.

In general, we want to use pipes in cases where writing foo4(foo3(foo2(foo1(data)))) becomes too tedious. Instead we could write it in the form data | foo1 | food2 | foo3

Here is one use case to illustrate its power:

from result import Result
import checkpipe as pipe
from typing import Tuple

print(
    [(4, 1, 3), (3, 2, 1), (10, 5, 5), (1, 3, 0)]
        | pipe.OfIter[Tuple[int, int, int]]
        .check(pipe.tup3_unpack(lambda n, m, sub_eq:
            n - m == sub_eq
        ))
        | pipe.OfIter[Result[Tuple[int, int, int], Tuple[int, int, int]]]
        .to_list()
)

[Ok((4, 1, 3)), Ok((3, 2, 1)), Ok((10, 5, 5)), Err((1, 3, 0))]

Here we are able to process a list of tuples, check a condition n - m == sub_eq against each of its elements, and finally consume that to a list. The check is preserved in whether the element ends up being tagged Ok or Err.

We can also ensure that all of the elements meet the condition using result flattening:

import checkpipe as pipe
from typing import Tuple

print(
    [(4, 1, 3), (3, 2, 1), (10, 5, 5), (1, 3, 0)]
        | pipe.OfIter[Tuple[int, int, int]]
        .check(pipe.tup3_unpack(lambda n, m, sub_eq:
            n - m == sub_eq
        ))
        | pipe.OfResultIter[Tuple[int, int, int], Tuple[int, int, int]]
        .flatten()
)

Err((1, 3, 0))

In this case it will give us a single tuple tagged error because not the entire list matched the property.

The | pipe.OfIter[Tuple[int, int ,int]] bit can be thought of as notating the type for the pipe so that the lambda can tell the types of its input. This can aid in intellisense and gives checkers like mypy full coverage. OfIter tells us that this applies to an Iterable of the specified type. There's also Of which applies to the type directly or OfResultIter which applies to an iterable of results, amongst others.

I have tried to see if we can specify the types by inference rather than manually, but I have not found a way to do so. If you have an idea, please reach out by opening an issue or PR.

And that's it! Please let me know what you think of this and any ideas you may have!

Thanks!
Lan

DEV Community: Lan

A New library for type-safe data pipelines in Python