philiprehberger-data-pipeline 0.1.0

Composable data transformation pipeline with lazy evaluation

philiprehberger-data-pipeline

Composable data transformation pipeline with lazy evaluation.

Install

pip install philiprehberger-data-pipeline

Usage

from philiprehberger_data_pipeline import Pipeline

data = [
    {"name": " Alice ", "email": "[e-mail]", "status": "active", "age": 30},
    {"name": "Bob", "email": "[e-mail]", "status": "inactive", "age": 25},
    {"name": "Alice", "email": "[e-mail]", "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

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)

Aggregations

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

Export

Pipeline(data).filter(...).to_csv("output.csv")
Pipeline(data).filter(...).to_json("output.json")

Operations

Transform	Description
.filter(fn)	Keep items where fn returns True
.map(fn)	Transform each item
.flat_map(fn)	Transform and flatten
.sort_by(key)	Sort by key (string or callable)
.unique_by(key)	Remove duplicates by key
.take(n)	Take first n items
.skip(n)	Skip first n items
.chunk(size)	Split into chunks
.flatten()	Flatten one level of nesting

Terminal	Description
.collect()	Execute and return list
.first()	Return first item
.count()	Count items
.sum(key)	Sum values
.avg(key)	Average values
.group_by(key)	Group into dict
.to_csv(path)	Export as CSV
.to_json(path)	Export as JSON

License

MIT