Skip to main content

kpipeline

kpipeline 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 (with the exception of certain async only pipes). 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
FallbackPipe Runs a fallback pipe in case the subpipe fails
AsyncBatchCollectorPipe Waits for a specific number of seconds, accumulating incoming data, and then processes them as a batch (async only, stateful)
AsyncSemaphorePipe Limits the number of concurrent calls to the subpipe (async only, stateful)
AsyncTimeoutPipe Cancels the subpipe after a certain amount of time (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 dislike writing tests so unit tests have been generated by Gemma 4 31B / Claude Opus 4.6 for your convenience. They are provided in the hope that they are better than nothing.

Metadata

Release files for kpipeline 1.4

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for kpipeline 1.4
File Size Uploaded Provenance Actions
kpipeline-1.4.tar.gz 15.2 kB Without provenance Details for kpipeline-1.4.tar.gz

Built distribution (wheel)

Table of built distributions (wheels) for kpipeline 1.4
File Interpreter ABI Platform Actions
kpipeline-1.4-py3-none-any.whl Python 3 none any Details for kpipeline-1.4-py3-none-any.whl

Total release size: 29.5 kB

Release files / kpipeline-1.4.tar.gz

Download URL kpipeline-1.4.tar.gz
Size 15.2 kB
Tags Source
SHA-256 checksum
How to use checksums
96f8000887f4e1b62d5842881d893f5e5a14a02b821473a6ef49d15e29cc50fa
BLAKE2b-256 checksum
How to use checksums
f257c325adac061f4eacc1a8e2bb6615ba1ab3caf003470a9c7d48099b2599f5
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via uv/0.11.32 {"installer":{"name":"uv","version":"0.11.32","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Fedora Linux Asahi Remix","version":"44","id":"","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}

Release files / kpipeline-1.4-py3-none-any.whl

Download URL kpipeline-1.4-py3-none-any.whl
Size 14.3 kB
Tags Python 3
SHA-256 checksum
How to use checksums
67d7fcaa9c66bbdb2e446741b5b200a3e674772914215be655c6edc8a0195a14
BLAKE2b-256 checksum
How to use checksums
5e44fa89212c8609e6ab02feef0b9d3d8c71c12b50b924f4836666680d2142d0
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via uv/0.11.32 {"installer":{"name":"uv","version":"0.11.32","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Fedora Linux Asahi Remix","version":"44","id":"","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}

Security for kpipeline 1.4

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

No provenance found

None of the files in this release were uploaded with provenance metadata.

A note on package security

PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.

To learn more about protecting yourself from malicious actors, visit our package security documentation.

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page