aiostream

0.7.1 · active · verified Fri Apr 10

aiostream provides a collection of stream operators that can be combined to create asynchronous pipelines of operations. It can be seen as an asynchronous version of itertools, although some aspects are are slightly different, offering features like operator pipe-lining, repeatability, and safe iteration contexts. The library is currently at version 0.7.1 and maintains an active development and release cadence.

Warnings

breaking In `aiostream` v0.6.0, stream operators (e.g., `map`, `filter`) transitioned from being classes to singleton objects. This means you should directly call `stream.map(...)` instead of `stream.map().pipe(...)` or `stream.Map().pipe(...)`.
Fix: Remove explicit instantiation of operators. For example, change `stream.map(func)` to `stream.map(source, func)` or `source | pipe.map(func)`.
breaking In `aiostream` v0.5.0, the `merge`, `chain`, and `ziplatest` operators no longer accept zero sources. Passing no arguments will now result in an error, as they expect at least one source stream to combine.
Fix: Ensure that `stream.merge()`, `stream.chain()`, or `stream.ziplatest()` are always provided with one or more asynchronous iterable sources.
gotcha Starting with `aiostream` v0.7.0, a `ValueError` is raised if `task_limit` or `ordered` arguments are provided to an operator (like `pipe.map`) when the supplied function is synchronous. These arguments are only applicable when the function being mapped is an asynchronous coroutine.
Fix: When using `pipe.map` or similar operators with a synchronous function, omit the `task_limit` and `ordered` arguments. Ensure these arguments are only used with asynchronous functions.
breaking `aiostream` dropped support for Python 3.8 in version 0.6.4. The library now officially requires Python 3.9 or newer.
Fix: Upgrade your Python environment to version 3.9 or later to use `aiostream` versions 0.6.4 and above.

Install

pip install aiostream Install latest version

Imports

stream
```
from aiostream import stream
```
The primary stream creation methods are exposed directly under 'aiostream.stream'.
pipe
```
from aiostream import pipe
```
Pipe operators are available directly from 'aiostream.pipe' for pipe-lining operations.

Quickstart

This quickstart demonstrates creating an asynchronous stream, applying transformations using pipe operators, iterating through the stream, and awaiting a stream to collect its results.

import asyncio
from aiostream import stream, pipe

async def main():
    # Create a counting stream with a 0.1-second interval
    xs = stream.count(interval=0.1)

    # Operators can be piped using '|' to transform the stream
    ys = xs \
        | pipe.map(lambda x: x**2) \
        | pipe.take(5) # Take the first 5 elements

    print("Squared stream elements:")
    # Use a stream context for proper resource management and iterate
    async with ys.stream() as streamer:
        async for y in streamer:
            print(f"-> {y}")

    # Streams can also be awaited to get the last value after processing
    last_value = await (stream.range(1, 4) | pipe.list())
    print(f"List from range stream: {last_value}")

if __name__ == "__main__":
    asyncio.run(main())

view raw JSON →