# aiostream, generator-based operators for asynchronous iteration

> aiostream gives Python async generators the operator vocabulary that itertools has for sync iterables, wrapped in a single stream interface that can be piped, sliced, concatenated and awaited. It is a pull-based library, closer to itertools than to a reactive framework.

**vxgmichel/aiostream** — Generator-based operators for asynchronous iteration

- Repository: https://github.com/vxgmichel/aiostream
- Website: http://aiostream.readthedocs.io
- Stars: 921 · Forks: 47
- Language: Python
- License: GPL-3.0
- Published: 2026-10-20 · Updated: 2026-10-20 · Language: en
- Canonical page: https://hysenlabs.com/projects/vxgmichel-aiostream

## It gives async generators the itertools vocabulary

aiostream provides a collection of stream operators that can be combined to create asynchronous pipelines of operations. The README describes it as an asynchronous version of itertools, while noting that some aspects are slightly different.

The central abstraction is the stream. Essentially all the provided operators return a unified interface called a stream, and a stream is an enhanced asynchronous iterable. That word enhanced is doing real work, because a stream adds six things to the plain async iterable protocol.

Operator pipe-lining using the pipe symbol. Transformation, selection and combination operators are applied with |, so a pipeline reads left to right in the source order.

Repeatability, where every iteration creates a different iterator. This is the property that distinguishes a stream from an ordinary async generator, and it is what makes a stream safe to await twice.

Safe iteration context, using async with and the stream method, so resources are released deterministically rather than relying on a generator being garbage collected.

Simplified execution, where awaiting a stream returns its last element.

Slicing and indexing with square brackets, and concatenation with the addition symbol.

The result is an interface that behaves like a collection of asynchronous values while being built from generator functions, which is a compromise rather than a reinvention of the async iteration protocol.

## The operator catalogue is organised by intent

The README groups the stream operators into seven categories, and the grouping tells you how the library thinks about the problem.

Creation covers the sources: iterate, preserve, just, call, empty, throw, never, repeat, count and range. Transformation covers map, enumerate, starmap, cycle and chunks. Selection covers take, takelast, skip, skiplast, getitem, filter, until, takewhile and dropwhile. Combination covers map, zip, merge, chain and ziplatest. Aggregation covers accumulate, reduce and list. Advanced covers concat, flatten, switch, concatmap, flatmap and switchmap. Timing covers spaceout, timeout and delay. Miscellaneous covers action and print.

Two things stand out in that list. The combination category includes map again, which points to the pipe module exposing variants of the same operator name for different pipeline positions. And the advanced category is where the interesting combinators live: concat and flatmap deal with streams of streams, while switch, switchmap and concatmap deal with the case where one source replaces another mid-pipeline, which is the situation that shows up when a websocket closes or a subscription is superseded.

The timing operators are the other practical group. A timeout operator bounds how long a stream may take, delay postpones a stream, and spaceout spreads items out over time. Those three cover most of what people hand-roll with asyncio.sleep in a polling loop.

## The demonstration shows the whole surface in one function

The README's example is compact enough to read in full, and it exercises most of the interface at once:

```python
import asyncio
from aiostream import stream, pipe


async def main():

    # Create a counting stream with a 0.2 seconds interval
    xs = stream.count(interval=0.2)

    # Operators can be piped using '|'
    ys = xs | pipe.map(lambda x: x**2)

    # Streams can be sliced
    zs = ys[1:10:2]

    # Use a stream context for proper resource management
    async with zs.stream() as streamer:

        # Asynchronous iteration
        async for z in streamer:

            # Print 1, 9, 25, 49 and 81
            print('->', z)

    # Streams can be awaited and return the last value
    print('9² = ', await zs)
```

Reading it in order tells you how the pieces fit. A counting stream produces increasing integers at a fixed interval. The map operator squares them. The slice takes elements 1 to 10 with a step of 2. The stream context is entered with async with, and inside it the async for loop consumes the values one at a time as they arrive.

The last line is the part that surprises people. Awaiting the stream returns its last value, 81, which is the ninth square in the sequence. That works because a stream is repeatable: awaiting it again produces the same last value from a fresh iteration rather than raising a StopAsyncIteration on the exhausted generator.

The demonstration also shows concatenation, since stream.just(1) + stream.range(2, 4) yields a stream of 1, 2 and 3, and the collection back to a list with await stream.list(one_two_three).

## Repeatability is the property everything else leans on

It is worth being precise about why repeatability matters, because it is the design decision that separates this library from a bag of async generator helpers.

An async generator is single-use. Once you have iterated it, it is exhausted and a second async for over it yields nothing. In a library built on plain generators, that constraint leaks into every operator, because an operator that consumes a source has to decide whether it is allowed to consume it again. Piping a filtered stream into a mapped stream and then reading the result twice is not something a generator-based design can offer without extra machinery.

A stream solves this by keeping the description of the pipeline rather than the single execution of it. Every iteration creates a different iterator, so each pass over a stream runs the underlying generators again. That is what lets the demonstration await zs twice and print the same result both times.

The consequence for the reader is that a stream is a recipe, not a value. Iterating it twice gives you two independent runs, with two independent timing, two independent network calls and two independent results. For a stream that hits a network endpoint, that is exactly what you want, and it is also a sharp edge, because nothing stops you from accidentally awaiting a stream that performs a side effect twice.

The safe iteration context exists for the same family of reasons. Entering async with zs.stream() gives you a streamer you can iterate, and leaving the block releases the underlying resources deterministically rather than waiting for a generator to be finalised.

## Where it sits between itertools and a reactive framework

The README links three references that frame the design: itertools, which is the synchronous ancestor, Rx, the reactive extensions model, and aioreactive, an earlier Python reactive library. It also links PEP 525, which is the Python enhancement proposal that made asynchronous generators and asynchronous comprehensions possible in the first place.

The distinction from a reactive framework is the one that matters in practice. aiostream is pull-based. A consumer asks for the next item, and the pipeline produces it, with the operators deciding when to pull from their sources. A push-based reactive library pushes items to subscribers as they become available, with buffering, scheduling and backpressure rules of its own.

For an async generator pipeline, pull is the natural model, since that is what async generators do, and PEP 525 is the mechanism underneath. If you need a subscriber that receives items pushed at whatever rate they arrive, with a scheduler deciding when they are processed, aiostream is not that library and adding it would not give you that.

The comparison with itertools is the closer one, and it holds with the caveat the README itself gives, that some aspects are slightly different. Most of the operator names map onto their itertools counterparts, which means the mental model transfers, and the differences are concentrated in the timing and combination operators where asyncio semantics, rather than sync semantics, decide the behaviour.

## What aiostream does not give you

The library is focused, and the gaps are worth stating before you plan an adoption around it.

It is an iteration library, not a scheduler. There is no event loop management, no task group, no structured concurrency layer and no retry policy. If your pipeline needs to run several sources concurrently and cancel them together, you are still writing that with asyncio primitives. The combination operators merge and zip streams, which is a data concern, but they are not a lifetime manager for the tasks behind those streams.

There is also no typed operator protocol of the kind a newer framework would offer. The project does care about typing, with mypy configured in strict mode for the aiostream package and for the examples, and pyright in strict mode for a specific list of modules including core.py, pipe.py and manager.py. The one runtime dependency is typing-extensions, and the package supports Python 3.10 through 3.14 with the requires-python floor set at 3.10.

The licensing is the other practical consideration. The project metadata says GPL-3.0-or-later, which is unusual for a small pure-Python utility. If you are pip installing it into a service, that is fine. If you plan to vendor it into a closed-source product, that is a conversation to have with your legal team before, not after.

Maintenance-wise the project is moving. The releases are frequent, with v0.8.1 on 2026-09-16, v0.8.0 on 2026-09-14 and v0.7.2 on 2026-09-08, and the last push to the repository was on 2026-10-09. A library on a 0.x version line can still change its API, so pin your dependency.

## Conclusion

aiostream suits a Python codebase that already uses async generators and wants a combinator library over them, or that wants to move from hand-written asyncio queues to a declarative pipeline. It does not suit a team that needs push-based reactive semantics, backpressure from a subscriber, or a framework with its own scheduler and operator ecosystem. Before you adopt it, confirm your Python version meets the requires-python floor of 3.10, and note that the licence is GPL-3.0-or-later, which matters if you intend to ship it inside a closed-source application.

## FAQ

### What is aiostream and what does it provide?

aiostream is a Python library of stream operators for asynchronous iteration, an asynchronous counterpart to itertools. All its operators return a unified stream interface, an enhanced asynchronous iterable that supports pipe-lining with |, repeatability, a safe iteration context with async with and stream(), awaiting for the last element, slicing and concatenation with +.

### How do I install aiostream?

The README states that you can install aiostream from PyPI as the aiostream package. The project metadata sets requires-python to 3.10 or later, and the only runtime dependency is typing-extensions.

### Why can an aiostream stream be iterated more than once?

Every iteration creates a different iterator, which is the repeatability property the README lists among the stream features. That is why awaiting the same stream twice returns the same last element instead of raising StopAsyncIteration, and it is also why a stream that has side effects will perform them once per iteration.

### What categories of operators does aiostream provide?

The README groups them into creation, transformation, selection, combination, aggregation, advanced, timing and miscellaneous. The timing group is spaceout, timeout and delay, and the advanced group covers concat, flatten, switch, concatmap, flatmap and switchmap.

### Is aiostream a reactive framework?

No. The README describes it as an asynchronous version of itertools, and it is pull-based, built on the asynchronous generator machinery of PEP 525, while it links Rx and aioreactive as related references. There is no scheduler, no task group and no push-based subscriber model.

## Sources

- [License: GPL-3.0](https://github.com/vxgmichel/aiostream/blob/main/LICENSE)
- [Project website](http://aiostream.readthedocs.io)
- [README](https://github.com/vxgmichel/aiostream/blob/main/README.md)
- [Releases](https://github.com/vxgmichel/aiostream/releases)
- [vxgmichel/aiostream on GitHub](https://github.com/vxgmichel/aiostream)

---

Hysen Labs editorial analysis, written from the project's own repository and release notes. Cite the canonical page: https://hysenlabs.com/projects/vxgmichel-aiostream