checkpipe

checkpipe

To bring functional programming data pipelines with robust validation to python and mypy

Why checkpipe? | Use Cases | Install | Todo | Sponsorship | Credits | Contributing | License

_{Built with ❤︎ by Mohammed Alzakariya}

Why checkpipe?

One problem is trying to express python functions in term of dataflows. Think of a function that progresses in stages like the following:

source_input -> filter -> transform -> filter -> sum

Dataflows can be more naturally represented with infix notation, with the preceding stage leading to the following stage through chaining. But in python we would find ourselves writing

sum(filter(transform(filter(source_input))))

which is not very handy. Another approach would be creating placeholder variables to store each stage, but this also introduces unnecessary state which goes against the principle of purity in functional programming.

In data analysis and ETL contexts, we may have to build large dataflows so a better approach is necessary.

a major inspiration for this project and a project which solves the above problem is pipe by Julien Palard. It allows for infix notation and gives a simple @Pipe decorator to extend this to any functionality the user needs.

This project aims to build on Julien Palard's project, but with new design considerations:

• Intellisense and mypy friendly: No untyped lambdas, get full autocomplete ability and type checking by mypy.
• Extended built-in support for error-checking which is integrated into the dataflow. This integrates with the rustedpy/Result which brings Rust-like Result[T, E] into python.

The project aims to make it easier to write pure python functions with robust error-checking and all the benefits of static analysis tools like mypy.

Use Cases

• Basic filtering and mapping
• Direct transformations outside iterators
• Basic validation in dataflows
• Creating a new Pipe function

Basic filtering and mapping

import checkpipe as pipe

print(
    [1, 2, 3]
    
    | pipe.OfIter[int].map(lambda n: n * 2)
    | pipe.OfIter[int].filter(lambda n: n != 4)
    | pipe.OfIter[int].to_list()
)

[2, 6]

The above example takes a source input [1, 2, 3] and transforms it by multiplying each value by 2 into, then keeping only results that aren't 4 and finally consuming this lazy iterator chain into a list result.

When using checkpipe, we are relying on specifying the type of the source in order for our lambdas to be typed. [1, 2, 3] is a List[int] and also can be iterated through as an Iterable[int]. Working with this type of source, we use pipe.OfIter[int]. This makes use of generics to give us expectations on the signature of the higher order functions passed to functions like .map and .filter. These expectations can be automatically checked by mypy. And vscode is able to know that n is an integer in the lambdas.

Direct transformations outside iterators

import checkpipe as pipe

print(
    3
    | pipe.Of[int].to(lambda n: n+1)
)

4

checkpipe does not only work with iterators. It works directly with types and allows transformations to the source object as well. In this case, no consumption of an iterator is jnecessary. .to(...) will return the transformed source directly.

Basic validation in dataflows

import checkpipe as pipe
from result import Result

print(
    [1, 2, 3]
    
    | pipe.OfIter[int].map(lambda n: n * 2)
    | pipe.OfIter[int].check(lambda n: n != 4)
    | pipe.OfIter[Result[int, int]].to_list()
)

[Ok(2), Err(4), Ok(6)]

Here, we are able to use .OfIter[int].check to apply a tag on all values in the source. Ok[int] when they pass the check n != 4 otherwise Err[int]. This allows us to propogate errors and handle errors in the pipeline itself. Note that when we're consuming the iterator pipeline with .to_list(), we are referring to a new source Iterator[Result[int, int]] to reflect the Ok/Err tagging.

We can now proceed to perform more computations on the Ok[int] results only:

import checkpipe as pipe
from result import Result

print(
    [1, 2, 3]
    
    | pipe.OfIter[int].map(lambda n: n * 2)
    | pipe.OfIter[int].check(lambda n: n != 4)
    | pipe.OfResultIter[int, int].on_ok(lambda n: n + 1)
    | pipe.OfIter[Result[int, int]].to_list()
)

[Ok(3), Err(4), Ok(7)]

Here, .OfResultIter[int, int] works with an iterable of Results as a source, and only when it detects an Ok, it performs the computation n+1. So we can see that Ok(2) became Ok(3) and Ok(6) became Ok(7), but Err(4) remains untouched.

We can also use a different type for the error:

import checkpipe as pipe
from result import Result

print(
    [1, 2, 3, 4]
    
    | pipe.OfIter[int].map(lambda n: n + 2)
    | pipe.OfResultIter[int, str].check(
        lambda n: n % 2 != 0,
        lambda n: f'Evens like {n} are not allowd!')
    | pipe.OfIter[Result[int, str]].to_list()
)

[Ok(3), Err('Evens like 4 are not allowd!'), Ok(5), Err('Evens like 6 are not allowd!')]

Here OfResultIter[int, str] specifies that errors will be in type str and Ok is in type int. It takes two functions, a predicate to check if the int is okay, and a function that maps from that int to some error message. We can then continue processing on just the Ok[int] results with .on_ok(...) just like before:

import checkpipe as pipe
from result import Result

print(
    [1, 2, 3, 4]
    
    | pipe.OfIter[int].map(lambda n: n + 2)
    | pipe.OfResultIter[int, str].check(
        lambda n: n % 2 != 0,
        lambda n: f'Evens like {n} are not allowd!')
    | pipe.OfResultIter[int, str].on_ok(lambda n: n * 10)
    | pipe.OfIter[Result[int, str]].to_list()
)

[Ok(30), Err('Evens like 4 are not allowd!'), Ok(50), Err('Evens like 6 are not allowd!')]

We can also chain multiple checks in a row, keeping in mind that checks on Result[T, E] use the then_check variants while checks on T use check.

import checkpipe as pipe
from result import Result

print(
    [1, 2, 3, 4]
    
    | pipe.OfIter[int].map(lambda n: n + 2)
    | pipe.OfResultIter[int, str].check(
        lambda n: n % 2 != 0,
        lambda n: f'Evens like {n} are not allowd!')
    | pipe.OfResultIter[int, str].then_check(
        lambda n: n != 3,
        lambda _: 'The number 3 is specifically not welcome!')
    | pipe.OfResultIter[int, str].on_ok(lambda n: n * 10)
    | pipe.OfIter[Result[int, str]].to_list()
)

[Err('The number 3 is specifically not welcome!'), Err('Evens like 4 are not allowd!'), Ok(50), Err('Evens like 6 are not allowd!')]

Sometimes the check itself reveals to us a property of the object that we want to error on. Instead of the error function taking the same object as the predicate, it gets a transformed object:

Unpacking tuples

checkpipe comes with support for unpacking tuples of limited size while specifying the types of each element:

import checkpipe as pipe

print(
    (4, 2, 'Hello ')
    | pipe.OfUnpack3[int, int, str].unpack(
        lambda num_spaces, repeat, text: '"' + ' ' * num_spaces + repeat * text + '"'
  )
)

"    Hello Hello "

Creating a new Pipe function

import checkpipe as pipe
from checkpipe import Pipe
from typing import Callable, Iterable

@Pipe
def multiply_by_num(num: int) -> Callable[[Iterable[int]], Iterable[int]]:
    def inner(source: Iterable[int]) -> Iterable[int]:
        return map(lambda n: n * num, source)
    return inner

print(
    [1, 2, 3]
    | multiply_by_num(3)
    | pipe.OfIter[int].to_list()
)

[3, 6, 9]

Here we create a new function that could utilize the pipe operator |, multiply_by_num. It defines an inner function which takes a source, Iterable[int], and it maps it to another Iterable[int] via the builtin map function.

If we want to utilize generics to create a more type-general pipe function, we could use typevars to infer types from the arguments passed into the function. If we want to inform the function about a more generic source type, we can wrap it in a class then inform of it the expected source type through the class like this:

import checkpipe as pipe
from checkpipe import Pipe
from typing import Generic, TypeVar, Callable, Iterable

T = TypeVar('T')

class Repeat(Generic[T]):
    @Pipe
    @staticmethod
    def repeat(n: int) -> Callable[[Iterable[T]], Iterable[T]]:
        def inner(source: Iterable[T]) -> Iterable[T]:
            for item in source:
                for _ in range(n):
                    yield item
        return inner

print(
    ['a', 'b', 'c']
    | Repeat[str].repeat(3)
    | pipe.OfIter[str].to_list()
)

['a', 'a', 'a', 'b', 'b', 'b', 'c', 'c', 'c']

The pipes are type-safe and they can be checked by mypy. checkpipe cannot automatically infer the source type from the left of the |. By specifiying Repeat[str], mypy knows that when the source ['a', 'b', 'c'] is piped to Repeat, that it must comply to being an Iterable[str] or mypy will error.

Install

pip install checkpipe

Todo

• Implement similar default pipes to Julien Palard's project to facilitate transition
• Implement unit testing for all functions of this module

Sponsorship

TODO

🎉 Credits

• Thanks to Julien Palard for the pipe library which was a major inspiration for this project.
• Thanks to jeffreystse for the README style.

Contributing

All contributions are welcome! I would appreciate feedback on improving the library and optimizing for use cases I haven't thought of yet! Please feel free to contact me by opening an issue ticket or emailing [e-mail] if you want to chat.

License

This theme is licensed under the MIT license © Mohammed Alzakariya.