A lightweight pipeline framework
Project description
PipeLayer
PipeLayer is a lightweight Python pipeline framework. Define a series of steps, and chain them together to create modular applications.
Table of Contents
Installation
From the command line:
pip install pipelayer
Getting Started
Step 1: Create Pipeline Filters
hello_world_filters.py
from pipelayer import Filter
class HelloFilter(Filter):
def run(self, data, context):
return "Hello"
class WorldFilter(Filter):
def run(self, data, context):
return f"{data}, World!"
functions.py
def create_message_dict(data, context):
return {"message": data}
Step 2: Create a Pipeline
Create a module to run the pipeline:
app.py
from pipelayer import Pipeline
from functions import create_message
from hello_world_filters import HelloFilter, WorldFilter
if __name__ = "__main__":
hello_world_pipeline = Pipeline([
HelloFilter, # pipeline.Filter type
WorldFilter, # pipeline.Filter instance
create_message_dict # function type
lambda data, context: json.dumps(data) # anonymous function
])
output = hello_world_pipeline.run()
# output = '{"message": "Hello, World!"}'
print(f"Pipeline Output: {output}")
print(hello_world_pipeline.manifest.__dict__)
Step 3: Run the Pipeline
from the command line:
run app.py
The Framework
pipelayer.Pipeline
__init__(steps, name)
args:
-
steps: List[Union[Step, Callable[[Any, Context], Any]]]
A list of:-
Classes and Instances that derive from
pipelayer.Filter
and implement therun
method -
Classes that implement the
pipelayer.Step
protocol -
Functions (instance/class/static/module) that have the following signature
def func(data: Any, context: Any)
-
Anonymous functions (lambda) with two arguments that follow this pattern:
my_func = lambda data, context: data
-
Instances of
pipelayer.Pipeline
-
-
name: Optional[str]
If not specified, the class name will be used.
Properties:
manifest: Manifest
An instance of pipelayer.Manifest
that is created when the run method is called.
Methods:
run(data, context) -> Any
The pipeline runner that iterates through the steps
and pipes filter output to the next step.
args:
data: Any
context: pipelayer.Context
pipelayer.Filter
__init__(name, pre_process, post_process)
args:
name: Optional[str]
If not specified, the class name will be used.pre_process: Optional[Callable[[Any, Context], Any]
post_process: Optional[Callable[[Any, Context], Any]
Properties:
pre_process: Optional[Callable[[Any, Context], Any]
post_process: Optional[Callable[[Any, Context], Any]
pipelayer.Step(Protocol)
Methods:
run(data, context) -> Any
args:
data: Any
context: pipelayer.Context
pipelayer.Context
A abstract base class for runtime app config.
pipelayer.Manifest
The Manifest keeps a record of Pipeline
and Filter
activity.
Utilities
pipelayer.util.render_manifest(manifest, indent) -> str
Static function that renders formatted JSON data
args:
manifest: Manifest
indent: Optional[int]
Default value is 2.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for pipelayer-0.3.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | c0f25279b8cda034ed5f81fa46ef7f26130ed37a10a845af55ce566bef74623a |
|
MD5 | 6759026522fabf22599b21401d661927 |
|
BLAKE2b-256 | 3740c674cb1d7e0995c892138017b2461389fc7396dd4c8bfbf9f4982bdde4ea |