scalable-pypeline 3.1.3

______ __   ________  _____  _     _____  _   _  _____
| ___ \\ \ / /| ___ \|  ___|| |   |_   _|| \ | ||  ___|
| |_/ / \ V / | |_/ /| |__  | |     | |  |  \| || |__
|  __/   \ /  |  __/ |  __| | |     | |  | . ` ||  __|
| |      | |  | |    | |___ | |_____| |_ | |\  || |___
\_|      \_/  \_|    \____/ \_____/\___/ \_| \_/\____/

PypeLine is a Python library for building and running distributed data pipelines. It provides:

• DAG Pipelines — define tasks as a directed acyclic graph; PypeLine handles execution order and parallelism
• Scheduled Tasks — run recurring jobs via cron schedule
• Flask API — built-in REST API with OpenAPI/Swagger documentation for triggering and managing pipelines
• Job Runner — one-shot worker for running a single pipeline task (e.g., in a Kubernetes Job)

PypeLine uses Dramatiq for task queuing, RabbitMQ (or Redis) as the message broker, and Redis for result storage.

Table of Contents

• Requirements
• Installation
• Project Structure
• 1. Define Your Tasks
• 2. Configure pypeline.yaml
• 3. Set Up Your Flask App
• 4. Set Environment Variables
• 5. Run the Services
• Configuration Reference
• API Authentication
• Testing

Requirements

• Python 3.8+
• RabbitMQ (default broker) or Redis (as broker)
• Redis (for result storage — always required)

Docker Compose is the easiest way to run RabbitMQ and Redis locally. See Run the Services below.

Installation

Install with the extras you need:

# Full install (Flask API + web server + pipeline workers)
pip install scalable-pypeline[flask,web,workers]

# Workers only (no Flask API)
pip install scalable-pypeline[workers]

Available extras:

Project Structure

PypeLine expects your application to be a Python package with a pypeline.yaml file inside it:

my_app/
├── my_app/
│   ├── __init__.py          # must define __version__
│   ├── pypeline.yaml        # PypeLine configuration
│   ├── pipeline.py          # your pipeline task functions
│   └── scheduled_tasks.py   # your scheduled task functions
├── app.py                   # Flask application factory
├── extensions.py            # Dramatiq extension
└── my_app.env               # environment variables

1. Define Your Tasks

Tasks are plain Python functions. Pipeline tasks receive an event argument containing the pipeline's input data. Scheduled tasks take no arguments.

my_app/pipeline.py

def step_a(event):
    # event contains the data passed when the pipeline was triggered
    data = event.get("my_input")
    print(f"Step A processing: {data}")


def step_b(event):
    print("Step B running (depends on A)")


def step_c(event):
    print("Step C running (depends on A, runs in parallel with B)")

my_app/scheduled_tasks.py

def hourly_report():
    print("Running hourly report...")

2. Configure pypeline.yaml

Place pypeline.yaml inside your Python package directory (next to __init__.py).

There are two pipeline schema versions. Use schemaVersion: 1 for simple pipelines and schemaVersion: 2 when you need multiple handlers per task node or typed pipeline settings.

Schema Version 1

Each task node maps to a single handler function. This is the simplest option and covers most use cases.

serviceConfig:
    - name: pipeline-worker
      registeredTasks:
          - handler: my_app.pipeline.step_a
          - handler: my_app.pipeline.step_b
          - handler: my_app.pipeline.step_c
          - handler: my_app.scheduled_tasks.hourly_report

pipelines:
    my-pipeline:
        name: My Pipeline
        description: An example pipeline with parallel steps
        schemaVersion: 1
        config:
            # DAG adjacency: step_a runs first, then step_b and step_c in parallel
            dagAdjacency:
                step_a:
                    - step_b
                    - step_c
            metadata:
                maxRetry: 3
                retryBackoff: 60        # seconds before first retry
                retryBackoffMax: 300    # max backoff seconds
                retryJitter: true
                maxTtl: 3600           # task time-to-live in seconds
                queue: pipeline-queue
            taskDefinitions:
                step_a:
                    handler: my_app.pipeline.step_a
                step_b:
                    handler: my_app.pipeline.step_b
                    queue: step-b-queue  # optional per-task queue override
                step_c:
                    handler: my_app.pipeline.step_c

scheduledTasks:
    hourly-report:
        name: Hourly Report
        enabled: true
        schemaVersion: 1
        config:
            task: my_app.scheduled_tasks.hourly_report
            queue: pipeline-queue
            schedule:
                minute: '0'
                hour: '*'
                dayOfWeek: '*'
                dayOfMonth: '*'
                monthOfYear: '*'

Schema Version 2

Version 2 adds two capabilities:

1. Multiple handlers per task node (handlers list instead of handler string). When triggering a pipeline you can specify which handler index to use for each node — this enables scenario-based execution (e.g. run an alternate algorithm for a given task without changing the pipeline structure).
2. Typed settings schema — define and validate the input parameters your pipeline accepts.

pipelines:
    my-pipeline-v2:
        name: My Pipeline V2
        description: Pipeline with swappable task handlers and validated settings
        schemaVersion: 2
        config:
            dagAdjacency:
                step_a:
                    - step_b
                    - step_c
            metadata:
                maxRetry: 3
                maxTtl: 3600
                queue: pipeline-queue
            taskDefinitions:
                step_a:
                    # Index 0: default handler; index 1: alternate implementation
                    handlers:
                        - my_app.pipeline.step_a_default
                        - my_app.pipeline.step_a_alternate
                step_b:
                    handlers:
                        - my_app.pipeline.step_b
                step_c:
                    handlers:
                        - my_app.pipeline.step_c

            # Optional: define typed, validated input settings for this pipeline
            settings:
                required:
                    - threshold
                properties:
                    threshold:
                        dataType: float
                        inputType: text
                        label: Decision Threshold
                        minimum: 0.0
                        maximum: 1.0
                    mode:
                        dataType: string
                        inputType: dropdown
                        label: Processing Mode
                        options:
                            - label: Fast
                              value: fast
                            - label: Accurate
                              value: accurate

All registered handlers must still be listed under serviceConfig.registeredTasks:

serviceConfig:
    - name: pipeline-worker
      registeredTasks:
          - handler: my_app.pipeline.step_a_default
          - handler: my_app.pipeline.step_a_alternate
          - handler: my_app.pipeline.step_b
          - handler: my_app.pipeline.step_c

Environment variable interpolation — use ${VAR_NAME} or ${VAR_NAME:default} anywhere in pypeline.yaml:

metadata:
    queue: ${TASK_QUEUE:pipeline-queue}

3. Set Up Your Flask App

extensions.py — create the Dramatiq extension:

from pypeline.dramatiq import Dramatiq

dramatiq = Dramatiq()

app.py — create the Flask application factory:

from flask import Flask
from pypeline.flask import FlaskPypeline
from extensions import dramatiq


def create_app():
    app = Flask(__name__)
    app.config.from_envvar("APP_CONFIG")  # or app.config.from_object(...)

    # Initialize Dramatiq broker (must happen before FlaskPypeline)
    dramatiq.init_app(app)

    # Initialize PypeLine — pass init_api=True to enable the REST API + Swagger UI
    pypeline = FlaskPypeline()
    pypeline.init_app(app, init_api=True)

    # Optionally register your own API blueprints into the PypeLine API
    # app.extensions["pypeline_core_api"].register_blueprint(my_bp)

    return app


if __name__ == "__main__":
    app = create_app()
    app.run(port=5001)

When init_api=True, PypeLine registers:

• GET/POST /api/v1/pipelines — list and trigger pipelines
• GET/POST /api/v1/schedules — list and manage scheduled tasks
• Swagger UI at /api/v1/docs

4. Set Environment Variables

# Required: tells PypeLine which package contains your pypeline.yaml
PYPELINE_CLIENT_PKG_NAME=my_app

# Redis URL (used for result storage, always required)
REDIS_URL=redis://:password@localhost:6379/0

# Message broker: RABBITMQ (default) or REDIS
MESSAGE_BROKER=RABBITMQ
RABBIT_URL=amqp://admin:password@localhost:5672

# Optional: protect API endpoints with an access key
API_ACCESS_KEY=your-secret-key

# Optional: override which worker config to load (defaults to WORKER_NAME env var)
# WORKER_NAME=pipeline-worker

5. Run the Services

Start Infrastructure

# docker-compose.yml (example)
# services: rabbitmq, redis
docker compose up -d

Run the Flask API

# Development
flask --app app run --port 5001

# Production
gunicorn "app:create_app()" --bind 0.0.0.0:5001 --worker-class gevent

Run Pipeline Workers

Workers consume tasks from the queues defined in pypeline.yaml:

flask --app app pypeline-worker

# Options:
#   -p / --processes   number of worker processes (default: CPU count)
#   -t / --threads     threads per process (default: 8)
#   -Q / --queues      comma-separated list of queues to consume from
flask --app app pypeline-worker -p 2 -t 4 -Q pipeline-queue

Run the Cron Scheduler

The scheduler reads your scheduledTasks config and enqueues tasks on their cron schedules:

flask --app app cron-scheduler

Run a One-Shot Job (Kubernetes Jobs)

The job-runner CLI starts a single worker that processes one task and exits — useful for Kubernetes Jobs:

# Listen on a specific queue; exits after processing one task
job-runner -q pipeline-queue

# With an idle timeout (exit if no job arrives within 30 seconds)
job-runner -q pipeline-queue --idle-timeout-ms 30000

Configuration Reference

pypeline.yaml Structure

Pipeline config Fields

Environment Variables

API Authentication

Protect your API by setting API_ACCESS_KEY in the environment. Clients must include the key in requests:

curl -H "accesskey: your-secret-key" http://localhost:5001/api/v1/pipelines

You can also use the require_accesskey decorator on your own endpoints:

from pypeline.flask.decorators import require_accesskey

@bp.route("/my-endpoint")
@require_accesskey
def my_endpoint():
    return {"status": "ok"}

Testing

Install test dependencies:

pip install -e ".[test]"

Run tests:

tox
# or directly with pytest
pytest