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 the run 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:

name: str

state: Pipeline.State

steps: List[Union[Step, Callable[[Any, Context], Any]]]

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.Switch

__init__(expression, cases, name)
An implementation of a Switch statement as a pipeline filter

args:

  • expression: Union[Step, Callable[[Any, Context], Any]]
  • cases: Dict[Union[Step, Callable[[Any, Context], Any]]]
  • name: Optional[str]
    If not specified, the class name will be used.

Properties:

expression: Union[Step, Callable[[Any, Context], Any]]
cases: Dict[Union[Step, Callable[[Any, Context], Any]]]
name: Optional[str]
manifest: Manifest

Methods:

run(data, context) -> Any
The switch runner that evaluates the specified expresssion executes the matching case.

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]

Events:
Events are lists of callables assignable after instantiation and are raised if the pipelayer.filter.raise_events decorator is applied to the implementation of the run method.

start: List[Callable[[Filter, Any], Any]]
Raised before the run method is invoked.

exit: List[Callable[[Filter, Any], Any]]
Raised if action is set to Action.SKIP or Action.EXIT in either a start or stop event handler.

end: List[Callable[[Filter, Any], Any]]
Raised after the run method is invoked.

pipelayer.FilterEventArgs

__init__(data, context, state)

args:

  • data: Any
  • context: Context
  • state: State

pipelayer.Context

A abstract base class for runtime app data.

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.

GitHub

https://github.com/greater-than/PipeLayer