Pipeline Class

from brokoli import Pipeline, source_db, transform, sink_file

with Pipeline("my-pipeline", description="Load and transform user data") as p:
    src = source_db("Load Users", conn_id="prod-pg", query="SELECT * FROM users")
    clean = transform("Clean", input=src, rules=[{"type": "drop_columns", "columns": ["password_hash"]}])
    out = sink_file("Export", input=clean, path="/data/users.csv", format="csv")

    src >> clean >> out

brokoli deploy

Pipeline(
    name: str,
    pipeline_id: str | None = None,
    description: str = "",
    schedule: str = "",
    catch_up: bool = False,
    sla: str = "",
    depends_on: list[str] | None = None,
    tags: list[str] | None = None,
    webhook: bool = False,
    max_retries: int = 0,
    concurrency: int = 4,
    on_start: Callable | None = None,
    on_success: Callable | None = None,
    on_failure: Callable | None = None,
)

with Pipeline("example") as p:
    # define nodes here
    ...

# After the block, p.to_yaml() returns readable YAML, p.to_json() returns a dict

from brokoli import Pipeline, source_db, transform, sink_file, sink_db

with Pipeline("fan-out") as p:
    src = source_db("Load", conn_id="pg", query="SELECT * FROM events")
    clean = transform("Clean", input=src, rules=[{"type": "rename", "old": "ts", "new": "timestamp"}])
    csv = sink_file("To CSV", input=clean, path="/data/events.csv")
    db = sink_db("To Warehouse", input=clean, table="events_clean", conn_id="warehouse")

    # Linear chain
    src >> clean

    # Fan-out from one node to multiple sinks
    clean >> csv
    clean >> db

src >> clean >> csv

from brokoli import source_db

src = source_db(
    "Load Users",
    query="SELECT * FROM users WHERE active = true",
    conn_id="prod-pg",       # reference a saved connection
    # uri="postgres://...",   # or pass a URI directly
    retries=3,
    retry_backoff="exponential",
    timeout=60,
)

from brokoli import source_api

api = source_api(
    "Fetch Orders",
    url="https://api.example.com/orders",
    method="GET",
    headers={"Authorization": "Bearer ${var.api_token}"},
    timeout=30,
)

from brokoli import source_file

f = source_file("Read CSV", path="/data/input.csv", format="csv")

from brokoli import transform

clean = transform(
    "Clean Data",
    input=src,
    rules=[
        {"type": "drop_columns", "columns": ["password_hash", "ssn"]},
        {"type": "rename", "old": "ts", "new": "timestamp"},
        {"type": "filter", "expression": "age >= 18"},
    ],
)

from brokoli import join

merged = join(
    "Merge Users + Orders",
    left=users,
    right=orders,
    on="user_id",
    how="left",
)

from brokoli import code

custom = code(
    "Custom Transform",
    input=src,
    language="python",
    script="df['full_name'] = df['first'] + ' ' + df['last']",
    python_path="/usr/bin/python3",
)

from brokoli import quality_check

qc = quality_check(
    "Validate",
    input=clean,
    rules=[
        {"type": "not_null", "columns": ["id", "email"]},
        {"type": "unique", "columns": ["id"]},
        {"type": "row_count", "min": 1},
    ],
)

from brokoli import condition_node

gate = condition_node(
    "Check Row Count",
    expression="row_count > 0",
    input=src,
)

from brokoli import sink_db

out = sink_db(
    "Write to Warehouse",
    input=clean,
    table="analytics.users",
    mode="append",
    conn_id="warehouse",
    retries=2,
)

from brokoli import sink_file

out = sink_file(
    "Export CSV",
    input=clean,
    path="/data/output.csv",
    format="csv",
    compress="gzip",
)

from brokoli import sink_api

out = sink_api(
    "Post Results",
    input=clean,
    url="https://api.example.com/ingest",
    method="POST",
    headers={"Content-Type": "application/json"},
)

from brokoli import migrate

m = migrate(
    "Replicate Users",
    source_conn_id="prod-pg",
    target_conn_id="warehouse",
    query="SELECT * FROM users",
    table="raw.users",
    mode="append",
)

from brokoli import dbt

d = dbt(
    "dbt Run",
    command="run",
    project_dir="/opt/dbt/my_project",
    target="prod",
    select="tag:daily",
)

from brokoli import notify

n = notify(
    "Alert on Complete",
    input=out,
    notify_type="webhook",
    webhook_url="https://hooks.slack.com/services/T.../B.../xxx",
    message="Pipeline finished successfully",
)

from brokoli import parallel, source_db, transform, sink_db

with Pipeline("parallel-sources") as p:
    users = source_db("Users", conn_id="pg", query="SELECT * FROM users")
    orders = source_db("Orders", conn_id="pg", query="SELECT * FROM orders")
    merge = transform("Merge", rules=[{"type": "concat"}])
    out = sink_db("Write", input=merge, table="combined", conn_id="warehouse")

    parallel(users, orders) >> merge >> out

with Pipeline("Daily User Export", pipeline_id="daily-user-export") as p:
    ...

from brokoli import Pipeline, source_db, sink_db

with Pipeline("daily-etl", schedule="0 5 * * *") as p:
    src = source_db("Load", query="SELECT *\nFROM users\nWHERE active = true", conn_id="pg")
    snk = sink_db("Save", table="output", conn_id="warehouse")
    src >> snk

print(p.to_yaml())

name: daily-etl
schedule: 0 5 * * *
enabled: true
nodes:
- id: load_a1b2c3
  type: source_db
  name: Load
  config:
    query: |
      SELECT *
      FROM users
      WHERE active = true
    conn_id: pg
- id: save_d4e5f6
  type: sink_db
  name: Save
  config:
    table: output
    mode: append
    conn_id: warehouse
edges:
- from: load_a1b2c3
  to: save_d4e5f6

import json
print(json.dumps(p.to_json(), indent=2))

def alert_on_failure(context):
    print(f"Pipeline {context['pipeline_name']} failed: {context['error']}")

with Pipeline(
    "critical-etl",
    schedule="0 6 * * *",
    on_failure=alert_on_failure,
    max_retries=2,
) as p:
    ...

from brokoli import (
    Pipeline,
    source_db,
    source_api,
    join,
    quality_check,
    transform,
    sink_db,
    notify,
)

with Pipeline(
    "user-enrichment",
    pipeline_id="user-enrichment",
    description="Enrich user profiles with latest activity data",
    schedule="0 */4 * * *",
    tags=["etl", "users"],
    max_retries=1,
    concurrency=4,
) as p:
    users = source_db(
        "Load Users",
        conn_id="prod-pg",
        query="SELECT id, email, name FROM users WHERE active = true",
        retries=2,
    )

    activity = source_api(
        "Fetch Activity",
        url="https://analytics.internal/api/v1/activity",
        headers={"Authorization": "Bearer ${var.analytics_token}"},
        timeout=60,
    )

    merged = join(
        "Join Users + Activity",
        left=users,
        right=activity,
        on="user_id",
        how="left",
    )

    qc = quality_check(
        "Validate",
        input=merged,
        rules=[
            {"type": "not_null", "columns": ["id", "email"]},
            {"type": "row_count", "min": 100},
        ],
    )

    clean = transform(
        "Normalize",
        input=qc,
        rules=[
            {"type": "rename", "old": "email", "new": "email_address"},
            {"type": "drop_columns", "columns": ["raw_json"]},
        ],
    )

    out = sink_db(
        "Write to Warehouse",
        input=clean,
        table="analytics.enriched_users",
        mode="append",
        conn_id="warehouse",
    )

    done = notify(
        "Slack Notification",
        input=out,
        notify_type="webhook",
        webhook_url="https://hooks.slack.com/services/T.../B.../xxx",
        message="User enrichment pipeline completed",
    )

    users >> merged
    activity >> merged
    merged >> qc >> clean >> out >> done

brokoli deploy