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)
| 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)
| File | Interpreter | ABI | Platform | Actions Reset |
|---|---|---|---|---|
| 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.