Metadata-Version: 2.4
Name: pipeline-toolkit
Version: 0.1.0
Summary: A small functional pipeline toolkit for Python.
Author-email: Hoàng Long <hoanglongcodes@gmail.com>
License-Expression: MIT
Project-URL: Homepage, https://github.com/Hoang-Long2012/pipeline-toolkit
Project-URL: Repository, https://github.com/Hoang-Long2012/pipeline-toolkit
Project-URL: Issues, https://github.com/Hoang-Long2012/pipeline-toolkit/issues
Project-URL: Changelog, https://github.com/Hoang-Long2012/pipeline-toolkit/blob/main/CHANGELOG.md
Keywords: pipeline,functional programming,functional pipeline,workflow,async,asynchronous,threading,utilities
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: 3.15
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Dynamic: license-file

# Pipeline Toolkit

A small functional pipeline toolkit for Python.

`pipeline-toolkit` provides a simple way to build sequential pipelines from ordinary Python callables. Each step receives the result of the previous step, while the pipeline runs asynchronously in a worker thread.

## Features

- Sequential functional pipeline execution
- Asynchronous execution using a worker thread
- Positional and keyword arguments for pipeline steps
- Stop, skip, wait, and rerun execution
- Manual synchronous step execution
- Result and error history using a stack
- Pipeline modification with `add()`, `insert()`, `pop()`, and `clear()`
- Small utility modules for functional workflows

## Installation

Install from PyPI:

```bash
pip install pipeline-toolkit
```

Or install directly from GitHub:

```bash
pip install git+https://github.com/Hoang-Long2012/pipeline-toolkit.git
```

## Quick Start

A pipeline is created from an iterable of steps. Each step is a tuple whose first item is a callable.

```python
from pipeline import Pipeline

def add(value, amount):
	return value + amount

def multiply(value, factor):
	return value * factor

pipeline = Pipeline([
	(add, (5,)),
	(multiply, (2,)),
])

pipeline.run(10).wait()

print(pipeline.results.get())
```

The execution flow is:

```text
10
 ↓
add(10, 5)
 ↓
15
 ↓
multiply(15, 2)
 ↓
30
```

The final result is `30`.

## Pipeline Steps

Each step can use one of four supported forms.

### Callable only

```python
(function,)
```

The callable receives the previous result:

```python
pipeline = Pipeline([
	(str.upper,),
])

pipeline.run("hello").wait()
```

### Positional arguments

```python
(function, args)
```

where `args` is a tuple:

```python
pipeline = Pipeline([
	(add, (5,)),
	(multiply, (2,)),
])
```

A step such as:

```python
(add, (5,))
```

is executed as:

```python
add(previous_result, 5)
```

### Keyword arguments

```python
(function, kwargs)
```

where `kwargs` is a mapping:

```python
pipeline = Pipeline([
	(pow, {"exp": 2}),
])
```

The step is executed as:

```python
pow(previous_result, exp=2)
```

### Positional and keyword arguments

```python
(function, args, kwargs)
```

For example:

```python
pipeline = Pipeline([
	(my_function, (1, 2), {"option": True}),
])
```

The callable receives the previous result followed by the supplied positional and keyword arguments.

## Execution

### `run()`

Start the pipeline asynchronously.

```python
pipeline.run(default=None, delay=0, daemon=False, stop_on_error=True)
```

The `default` value becomes the initial result and is passed to the first step.

`delay` specifies the delay between steps.

`daemon` controls whether the worker thread is a daemon thread.

`stop_on_error` controls whether execution stops after the first exception.

`run()` returns the pipeline instance, allowing calls such as:

```python
pipeline.run(10).wait()
```

The pipeline is snapshotted when execution starts. Changes made to `pipeline.pipeline` after `run()` begins do not affect the current execution.

### `wait()`

Wait for the current execution to finish.

```python
pipeline.wait()
```

It returns the pipeline instance.

### `stop()`

Request the running pipeline to stop and wait for its worker thread to terminate.

```python
step = pipeline.stop()
```

The return value is the current one-based step index when execution is stopped, or `0` if the pipeline was not running.

### `skip()`

Request the worker to skip the next step that reaches its skip check.

```python
pipeline.skip()
```

The method returns the pipeline instance.

### `rerun()`

Stop the current execution and start the pipeline again.

```python
pipeline.rerun(10)
```

Arguments are passed directly to `run()`.

## Manual Step Execution

`run_step()` executes one configured step synchronously.

```python
result = pipeline.run_step(2, 10)
```

Unlike `run()`, this method:

- does not create a worker thread
- does not modify the worker thread or pipeline execution state
- does not store the result in `results`
- does not store exceptions in `errors`
- allows exceptions to propagate to the caller

This makes it useful when a single pipeline step needs to be executed manually.

## Results and Errors

The pipeline provides two `Stack` instances:

```python
pipeline.results
pipeline.errors
```

`results` contains the initial value and the results produced by executed steps.

For example:

```python
pipeline.run(10).wait()

print(pipeline.results.get())
```

`errors` contains exceptions raised by pipeline steps.

When `stop_on_error=True`, execution stops after the first exception.

When `stop_on_error=False`, the exception is stored in `errors` and execution continues with the previous result.

## Managing Pipeline Steps

Pipeline steps can be modified before or between executions.

### `add()`

Append a step:

```python
pipeline.add((str.upper,))
```

### `insert()`

Insert a step at a one-based position:

```python
pipeline.insert(2, (str.strip,))
```

### `pop()`

Remove and return a step:

```python
step = pipeline.pop(1)
```

Pipeline indexes are one-based.

### `clear()`

Remove all configured steps:

```python
pipeline.clear()
```

## Pipeline State

The `running` property indicates whether the worker thread is currently running:

```python
if pipeline.running:
	print("Pipeline is running")
```

The `step` attribute contains the one-based index of the currently executing step. It is `0` when the pipeline is not running.

A `Pipeline` instance can also be used as a boolean:

```python
if pipeline:
	print("Pipeline is running")
```

Calling a pipeline instance is equivalent to calling `run()`:

```python
pipeline(10)
```

is equivalent to:

```python
pipeline.run(10)
```

The length of a pipeline is the number of configured steps:

```python
len(pipeline)
```

A callable can be checked with the `in` operator:

```python
if add in pipeline:
	print("add is part of the pipeline")
```

Callable membership uses identity comparison.

## Utilities

### `Stack`

`Stack` is a simple LIFO stack container with optional capacity limits.

It supports common stack operations such as pushing, retrieving, peeking, and removing items, with dedicated exceptions for overflow and underflow conditions.

Import it directly from its submodule:

```python
from pipeline.stack import Stack
```

`Stack` is also used internally by `Pipeline` for storing results and errors.

For example:

```python
from pipeline.stack import Stack

stack = Stack()

stack.push("first")
stack.push("second")

print(stack.get())
```

For detailed stack operations and behavior, see the `pipeline.stack` module.

### `tap`

`tap` is a small functional utility for performing a side effect while keeping the pipeline value available for subsequent processing.

`tap` performs a side effect on a deep copy of the current value and returns the original value unchanged.

Import it directly from its submodule:

```python
from pipeline.tap import tap
```

For example:

```python
from pipeline import Pipeline
from pipeline.tap import tap

def add(value, amount):
	return value + amount

pipeline = Pipeline([
	(add, (5,)),
	(tap, (print,)),
	(add, (10,)),
])

pipeline.run(10).wait()
```

Utilities are provided as separate submodules rather than being exported from the top-level `pipeline` package.

## API Overview

### `Pipeline`

| Member       | Description                           |
| ------------ | ------------------------------------- |
| `run()`      | Start asynchronous pipeline execution |
| `run_step()` | Execute one step synchronously        |
| `stop()`     | Stop the current execution            |
| `skip()`     | Request the next step to be skipped   |
| `wait()`     | Wait for the current execution        |
| `rerun()`    | Restart the pipeline                  |
| `add()`      | Append a step                         |
| `insert()`   | Insert a step                         |
| `pop()`      | Remove and return a step              |
| `clear()`    | Remove all steps                      |
| `running`    | Whether the worker is running         |
| `step`       | Current one-based step index          |
| `results`    | Stack of initial value and results    |
| `errors`     | Stack of raised exceptions            |

## Requirements

* Python 3.8 or newer

## Changelog

See changelog from: [CHANGELOG.md](https://github.com/Hoang-Long2012/pipeline-toolkit/blob/main/CHANGELOG.md)

## License

This project is licensed under the MIT License. See [LICENSE](https://github.com/Hoang-Long2012/pipeline-toolkit/blob/main/LICENSE) for details.

## Contribution

If you'd like to contribute, feel free to submit a pull request.  
If you'd like to report a bug or request a feature, please open an issue.

Copyright (C) 2026 Hoàng Long
