philiprehberger-data-pipeline

Composable data transformation pipeline with lazy evaluation.

Installation

pip install philiprehberger-data-pipeline

Usage

from philiprehberger_data_pipeline import Pipeline

data = [
    {"name": " Alice ", "email": "alice@example.com", "status": "active", "age": 30},
    {"name": "Bob", "email": "bob@example.com", "status": "inactive", "age": 25},
    {"name": "Alice", "email": "alice@example.com", "status": "active", "age": 30},
]

result = (
    Pipeline(data)
    .filter(lambda r: r["status"] == "active")
    .map(lambda r: {**r, "name": r["name"].strip()})
    .unique_by("email")
    .sort_by("name")
    .collect()
)

Reusable Pipelines

from philiprehberger_data_pipeline import Pipeline

clean_users = (
    Pipeline.define()
    .filter(lambda r: r.get("email"))
    .map(lambda r: {**r, "email": r["email"].lower()})
    .unique_by("email")
)

active = clean_users.run(active_users)
archived = clean_users.run(archived_users)

Tap (Side Effects)

from philiprehberger_data_pipeline import Pipeline

result = (
    Pipeline([1, 2, 3])
    .tap(lambda x: print(f"Processing: {x}"))
    .map(lambda x: x * 2)
    .collect()
)
# Prints each item without altering the data

Branch (Parallel Splits)

from philiprehberger_data_pipeline import Pipeline

result = (
    Pipeline([1, 2, 3])
    .branch(
        lambda p: p.map(lambda x: x * 2).collect(),
        lambda p: p.filter(lambda x: x > 1).collect(),
    )
    .collect()
)
# [2, 4, 6, 2, 3]

Retry Wrapper

from philiprehberger_data_pipeline import Pipeline, retry

def fetch_url(url):
    # might fail transiently
    return requests.get(url).text

result = Pipeline(urls).map(retry(fetch_url, attempts=3, delay=1.0)).collect()

Pipeline Composition

from philiprehberger_data_pipeline import Pipeline

clean = Pipeline.define().filter(lambda x: x > 0).map(lambda x: x * 2)
limit = Pipeline.define().take(3)

combined = clean + limit
combined.run([−1, 5, 0, 3, 7, 2])
# [10, 6, 14]

Dry Run

from philiprehberger_data_pipeline import Pipeline

log = (
    Pipeline([1, 2, 3, 4])
    .filter(lambda x: x > 2)
    .map(lambda x: x * 10)
    .dry_run()
)
# [{"step": 0, "name": "filter", "input": [1,2,3,4], "output": [3,4]},
#  {"step": 1, "name": "map", "input": [3,4], "output": [30,40]}]

Sliding Window

from philiprehberger_data_pipeline import Pipeline

Pipeline([1, 2, 3, 4, 5]).window(3, 1).collect()
# [[1, 2, 3], [2, 3, 4], [3, 4, 5]]

Aggregations

from philiprehberger_data_pipeline import Pipeline

p = Pipeline(sales_data)
total = p.sum("amount")
average = p.avg("amount")
grouped = p.group_by("category")

Peek and count_by

peek(n) materializes the first n items for quick inspection without committing to a collect(). count_by(key) returns a per-key occurrence count, similar to group_by but without the per-value lists.

from philiprehberger_data_pipeline import Pipeline

Pipeline([10, 20, 30, 40, 50, 60]).filter(lambda x: x > 15).peek(3)
# [20, 30, 40]

Pipeline([
    {"category": "fruit"},
    {"category": "veg"},
    {"category": "fruit"},
]).count_by("category")
# {"fruit": 2, "veg": 1}

Export

from philiprehberger_data_pipeline import Pipeline

Pipeline(data).filter(lambda x: x["active"]).to_csv("output.csv")
Pipeline(data).filter(lambda x: x["active"]).to_json("output.json")

API

Development

pip install -e .
python -m pytest tests/ -v

Support

If you find this project useful:

⭐ Star the repo

🐛 Report issues

💡 Suggest features

❤️ Sponsor development

🌐 All Open Source Projects

💻 GitHub Profile

🔗 LinkedIn Profile

License

MIT