kpipeline 1.1.1

A pipeline library for Python

kpipe

kpipe is a simple pipeline library that allows you to write your complex application as a pipeline. This helps code organization, debugging, and testing.

How to use

kpipe is like a DSL language inside Python. It is based on "pipes" which can be combined into pipelines using the pipe primitives provided by the library.

Each pipe represents a function transforming an input into an output. Pipes are stateless and immutable. Below is an example defining two example pipes.

from kpipeline import Pipe

class AddOnePipe(Pipe[int, int, None]):
    def apply(self, input: int, metadata: None) -> int:
        return input + 1


class MulByTwoPipe(Pipe[int, int, None]):
    def apply(self, input: int, metadata: None) -> int:
        return input * 2

These pipes can be combined using the ChainPipe primitive (aliased into the | operator) two form a pipeline that performs these two pipes sequentially:

pipeline = AddOnePipe() | MulByTwoPipe()

# or

from kpipeline import ChainPipe
pipeline = ChainPipe(AddOnePipe(), MulByTwoPipe())

print(pipeline.apply(2))  # 6

Currently, the library defines these primitives:

Primitive	Purpose
ChainPipe	Execute two pipes sequentially
ConditionalPipe	Execute a pipe if a given condition is true
BranchPipe	Execute one of two pipes depending on whether the condition is true or false
SelectPipe	Execute one of multiple pipes based on a selector key
ParallelPipe	Execute multiple pipes and combine their results (only async implementation is actually parallel)
MetadataWrapperPipe	Transform the metadata given to the pipeline for the subpipe
MapPipe	Apply a subpipe into a sequence of inputs
FilterPipe	Filter a sequence of inputs using a predicate
RetryPipe	Run a pipe multiple times in case it fails
AsyncBatchCollectorPipe	Waits for a specific number of seconds, accumulating incoming data, and then processes them as a batch (async only)

Tests

You can run all tests in this repository with

uv run pytest test
uv run mypy .

License

TBD

AI Use Disclosure

I hate writing tests so unit tests have been generated by Gemma 4 31B for your convenience. They are provided in the hope that they are better than nothing.