pipeline-frame 0.1.1

A lightweight in-memory pipeline framework built around before, run, and after.

pipeline-frame

pipeline-frame is a lightweight in-memory pipeline framework for Python 3.11+.

It keeps the model intentionally narrow: every step is built around before(), run(), and after().

Core idea

Define a linear pipeline with Step instances and | composition:

from pipeline_frame import Pipeline, PipelineContext, Step


class A(Step):
    def before(self, context: PipelineContext) -> None:
        print("a.before")

    def run(self, context: PipelineContext) -> None:
        context.set("text", context.input["text"].strip())

    def after(self, context: PipelineContext) -> None:
        print("a.after")


class B(Step):
    def run(self, context: PipelineContext) -> None:
        context.set_output(context.get("text").upper())


pipeline = A() | B()

result = pipeline.run({"text": "  hello world  "})
print(result.output)

Execution is linear and easy to reason about:

A.before()
A.run()
A.after()
B.before()
B.run()
B.after()

Async steps use the same lifecycle, but run through await pipeline.arun(...):

from pipeline_frame import Pipeline, PipelineContext, Step


class Fetch(Step):
    async def run(self, context: PipelineContext) -> None:
        context.set("text", context.input["text"].strip())


class Render(Step):
    async def run(self, context: PipelineContext) -> None:
        context.set_output(context.get("text").upper())


pipeline = Pipeline("request").add(Fetch).add(Render)
result = await pipeline.arun({"text": "  hello world  "})
print(result.output)

Pipeline.arun() accepts mixed sync/async step methods. If a step defines async lifecycle methods, do not call Pipeline.run().

If you want to keep an explicit pipeline name, you can also write:

pipeline = Pipeline("request") | A() | B()

Design goals

• one core abstraction: Pipeline
• one step lifecycle: before, run, after
• chainable definitions with minimal ceremony
• structured execution results
• predictable stop-on-failure behavior

Failure behavior

• if before() fails, the current step stops and later steps do not run
• if run() fails, the current step still attempts after()
• if after() fails, the pipeline is marked failed and later steps do not run
• all details are available on PipelineResult

API overview

• Pipeline: ordered collection of step instances
• Step: base class with before(), run(), and after()
• PipelineContext: shared input, state, output, and error
• PipelineResult: execution report for one run

Development

Run tests from the repository root:

python -m unittest discover -s tests -v