fpstreams

fpstreams

fpstreams is a Python library designed for processing iterables and records through lazy pipelines, enabling bounded asynchronous concurrency and efficient resource management. This project showcases an innovative approach to handling data transformations and aggregations in Python, allowing for rich, scalable data processing with ease.

Key Features:

• Synchronous and Asynchronous Pipelines: Use flow() for synchronous pipelines, or aflow() for asynchronous processing that respects concurrency limits.
• Flexible Data Handling: Utilize pipelines that work with values, records, and key/value pairs, supporting operations like filtering, mapping, and aggregation.
• Explicit Resource Management: Cleanup resources automatically with finally blocks to ensure that generators and iterators are properly closed upon completion.

Basic Usage:

Synchronous Example

Here's a simple example demonstrating how to use the flow() function:

from fpstreams import flow

result = flow([1, 2, 3]).filter(lambda value: value > 1).map(lambda value: value * 10).to_list()
assert result == [20, 30]

Asynchronous Example

For asynchronous processing, use the aflow() function:

import asyncio
from fpstreams import aflow

async def fetch(value: int) -> int:
    await asyncio.sleep(0.01)
    return value * 10

async def main() -> None:
    result = await (
        aflow([1, 2, 3, 4])
        .map_async(fetch, concurrency=2, ordered=True)
        .filter(lambda value: value >= 20)
        .to_list()
    )
    print(result)
    # Output: [20, 30, 40]
asyncio.run(main())

Advanced Features:

• Custom Aggregations: Create advanced aggregations by using the aggregate() function along with built-in aggregation functions such as count(), sum(), and mean().
• Record Pipelines: Use record pipelines to operate on data structures like dictionaries and named tuples efficiently, allowing for operations like group_by() and with_columns().
• Performance Optimization: Monitor and adjust execution using different execution engines, including Python, Rust, and Arrow, to optimize performance for different data workloads.

Example of Record Aggregation

Using flow() for record aggregation:

from fpstreams import agg, col, flow

orders = [
    {"region": "eu", "status": "paid", "price": 12, "quantity": 2},
    {"region": "us", "status": "paid", "price": 20, "quantity": 1},
    {"region": "eu", "status": "cancelled", "price": 99, "quantity": 1},
    {"region": "eu", "status": "paid", "price": 8, "quantity": 3},
]

revenue = (
    flow(orders)
    .filter(col("status") == "paid")
    .with_columns(revenue=col("price") * col("quantity"))
    .group_by("region")
    .aggregate(
        orders=agg.count(),
        revenue=agg.sum("revenue"),
    )
    .sort_by("region")
    .to_list()
)

assert revenue == [
    {"region": "eu", "orders": 2, "revenue": 48},
    {"region": "us", "orders": 1, "revenue": 20},
]  

In summary, fpstreams offers a robust and efficient way to handle data processing in Python, enhancing productivity with its lazy evaluation and resource-efficient operations. Explore the documentation to learn more about this powerful library.