pipelayer 0.3.0

A lightweight pipeline framework

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
• Getting Started
• The Framework
  
  

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

• Pipeline
• Filter
• Context
• Manifest
• Utilities
  
  

pipelayer.Pipeline(Step)

Constructor

__init__(self: Pipeline, steps: List[Union[Step, Callable[Any, [Context]]]] = None, name: str = "")
The type hints for the steps arg may look confusing. Here's what's allowed:

• Classes and Instances that derive from pipelayer.Filter and implement the run method

• Functions (instance/class/static/module) that have the following signature

  def func(data: Any, context: Any)
  

• Anonymous functions (lambda) with two arguments that follow the same pattern for regular functions:

  my_func = lambda data, context: data
  

• Instances of pipelayer.Pipeline (new in v0.3.0)

Properties

manifest (Manifest)
An instance of pipelayer.Manifest that is created when the run method is called.

Methods

run(data: Any, context: Optional[Context]) -> Any
The pipeline runner that iterates through the steps and pipes filter output to the next step.

pipelayer.Filter(Step)

A base class with an abstract run method.

Properties

pre_process (callable)
Optional.

post_process (callable)
Optional.

Methods

@abstractmethod
run(data: Any, context: Optional[Context]) -> Any
The abstract filter runner.

pipelayer.Step

The base class that is sub-classed by pipelayer.Pipeline and pipelayer.Filter.

Abstract Methods

@abstractmethod
run(data: Any, context: Optional[Context]) -> Any
The abstract

Properties

name (str)
Optional. Used by the Manifest

pipelayer.Context

A extensible base class for runtime app config.

pipelayer.Manifest

The Manifest keeps a record of Pipeline and Filter activity.

Utilities

pipelayer.util.render_manifest(manifest: Manifest, indent: int = 2) -> str Static function that renders formatted JSON data