philiprehberger-data-pipeline 0.3.1

Composable data transformation pipeline with lazy evaluation

philiprehberger-data-pipeline

Composable data transformation pipeline with lazy evaluation.

Installation

pip install philiprehberger-data-pipeline

Usage

Basic Pipeline

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)

Sliding Window

data = [1, 2, 3, 4, 5]

# Window of size 3, step 1
Pipeline(data).window(3, 1).collect()
# [[1, 2, 3], [2, 3, 4], [3, 4, 5]]

# Window of size 3, step 2
Pipeline(data).window(3, 2).collect()
# [[1, 2, 3], [3, 4, 5]]

Deduplication

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

# Works with unhashable items too
Pipeline([{"a": 1}, {"a": 1}, {"b": 2}]).deduplicate().collect()
# [{"a": 1}, {"b": 2}]

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")

API

Function / Class	Description
Pipeline(data)	Composable, lazy data transformation pipeline with chainable operations and terminal methods
.filter(fn)	Keep items where fn returns True
.map(fn)	Transform each item
.flat_map(fn)	Transform and flatten
.flatten()	Flatten one level of nesting
.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
.each(fn)	Execute side effect for each item
.window(size, step)	Sliding window grouping
.deduplicate()	Remove duplicate items preserving order
.collect()	Execute and return list
.first()	Return first item
.count()	Count items
.sum(key)	Sum values
.avg(key)	Average values
.min(key)	Find minimum value
.max(key)	Find maximum value
.reduce(fn, initial)	Reduce to single value
.group_by(key)	Group into dict
.to_csv(path)	Export as CSV
.to_json(path)	Export as JSON

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