pipelayer 0.1.0

A lightweight pipeline framework

PipeLayer

PipeLayer is a lightweight pipeline framework. Define a series of filters, and chain them together to create modular applications.

from app_context import AppContext
from app_settings import AppSettings
from pipelayer import Pipeline
from hello_filter import HelloFilter
from world_filter import WorldFilter
from logging import Logger

app_settings = AppSettings()
app_context = AppContext(app_settings, Logger("Logger"))
pipeline = Pipeline.create(app_context, "Hello World Pipeline")

output = pipeline.run([
    HelloFilter(),
    WorldFilter()
])

print(f"Pipeline Output: {output}")
print(pipeline.manifest.__dict__)

Table of Contents

• Installation
• Quick Start
• The Framework
• Examples
  
  

Installation

From the command line:

pip install pipelayer

Quick Start

Step 1: Application Settings

Create a class called AppSettings that inherits from the pipelayer.Settings class:

app_settings.py

from pipelayer import Settings


class AppSettings(Settings):
    """
    Complete this by adding constants, key/value data from AWS Parameter Store, etc

    The pipelayer.Settings class inherits from pydantic.BaseModel, so fields must be typed appropriately
    """
    ...

NOTE: You do not have to use the Settings base class in your application. Any class can be provided.

Step 2: Application Context

Create a class called AppContext that inherits from the pipelayer.Context class:

app_context.py

from logging import Logger

from app_settings import AppSettings
from pipelayer import Context


class AppContext(Context):
    def __init__(self, settings: AppSettings, log: Logger = Nones):
        self.__settings = settings
        self.__log = log

    @property
    def settings(self) -> AppSettings:
        return self.__settings

    @property
    def log(self) -> Logger:
        return self.__log

Step 3: Create Pipeline Filters

Create classes that inherits from pipelayer.Filter:

hello_filter.py

from pipelayer import Filter


class HelloFilter(Filter):
    def run(self, context) -> str:
        return "Hello"

world_filter.py

from pipelayer import Filter


class WorldFilter(Filter):
    def run(self, context, data) -> str:
        return data + " World!"

Step 4: Create a Pipeline

Create a module to run the pipeline:

app.py

from app_context import AppContext
from app_settings import AppSettings
from pipelayer import Pipeline
from hello_filter import HelloFilter
from world_filter import WorldFilter
from logging import Logger

app_settings = AppSettings()
app_context = AppContext(app_settings, Logger("Logger"))
hello_world_pipeline = Pipeline.create(app_context, "Hello World Pipeline")

output = hello_world_pipeline.run([
    HelloFilter(),
    WorldFilter()
])

print(f"Pipeline Output: {output}")
print(hello_world_pipeline.manifest.__dict__)

Step 5: Run the Pipeline

from the command line:

run app.py

---

The Framework

• Pipeline
• Filter
• Context
• Settings
• Manifest
  
  

pipelayer.Pipeline

Properties

name (str)
An optional name. It's used in the Manifest.

context (Context)
An instance of pipelayer.Context.

manifest (Manifest)
An instance of pipelayer.Manifest that is created at runtime.

Methods

create(context: Context, name: str = "") -> Pipeline
The factory method to create a pipeline

run(filters: List[Filter], data: Any = None) -> Any
The pipeline runner that iterates through the filters and pipes filter output to the next filter.

pipelayer.Filter

A functional unit that implements the run method, and the optional pre_process and post_process methods.

Properties

name (str)
Optional. Used by the Manifest

pre_process (callable)
Optional.

post_process (callable)
Optional.

Methods

run(context: Context, data: Any) -> Any
The type of the data argument in the abstract class is Any, but you can use the correct type for the data when implementing method. The same is true for the return type.

pipelayer.Context

The Context object is used to pass application-level information to each filter.

Properties

settings (Any)
An optional abstract property for storing application-level data.

log (Any)
An optional abstract property for a common logger that can be used within pipeline filters.

manifest (Manifest)
A model that keeps a record of start/stop times for a pipeline, as well as each filter.

pipelayer.Settings

The settings class is an optional base class for applications settings that inherits from pydantic.BaseModel.

pipelayer.Manifest

The Manifest keeps a record of Pipeline and Filter activity.