Streamlet

A fluent, lazy stream-processing library for Python, modeled after Java's Streams API.

from streamlet import Stream

(
    Stream.from_iterable(range(100))
    .filter(lambda n: n % 2 == 0)
    .map(lambda n: n**2)
    .take(5)
    .to_list()
)
# [0, 4, 16, 36, 64]

The goal is to make itertools' power reachable through clean, chainable syntax.

pip install pystreamlet    # or: uv add pystreamlet

You install pystreamlet but import streamlet, as above — PyPI reserves names confusable with streamlit. Requires Python 3.11+, has no dependencies, and ships type information (py.typed).

Status: 0.1.0, awaiting its first upload. Everything below works and is tested; until the release lands on PyPI, install from source (see Development). Before 1.0.0 the API may still change — see versioning.

Why

itertools is powerful but reads inside out — you write the last step first:

# itertools: read from the middle outwards
from itertools import count, islice

list(islice((n**2 for n in count() if n % 2 == 0), 5))

# streamlet: read left to right, in the order it happens
(
    Stream.iterate(0, lambda n: n + 1)
    .filter(lambda n: n % 2 == 0)
    .map(lambda n: n**2)
    .take(5)
    .to_list()
)

Both are lazy and both handle infinite sources. The difference is that one describes the pipeline in the order it runs.

Side by side

Take the first few results of an expensive step. islice wraps what it slices, so the bound you care about ends up furthest from the work it bounds:

from itertools import islice

list(islice(map(fetch, filter(is_recent, urls)), 5))

Stream(urls).filter(is_recent).map(fetch).take(5).to_list()

Group items by a key. itertools.groupby only groups adjacent items, so it needs a sort first and hands back iterators that expire as you advance:

from itertools import groupby
from operator import attrgetter

by_team = {
    team: list(members)
    for team, members in groupby(sorted(users, key=attrgetter("team")), key=attrgetter("team"))
}

by_team = Stream(users).group_by(attrgetter("team"))

Drop repeats but keep order. The set version loses order, and the dict.fromkeys trick reads as a puzzle:

list(dict.fromkeys(names))

Stream(names).distinct().to_list()

Chain a few steps over a file. The itertools version needs a with block, a generator expression and a slice, in three different directions:

with open("app.log", encoding="utf-8") as handle:
    first_errors = list(islice((line for line in handle if line.startswith("ERROR")), 10))

with Stream.from_file("app.log") as lines:
    first_errors = lines.filter(lambda line: line.startswith("ERROR")).take(10).to_list()

Stop as soon as one item matches. next() with a default over a generator expression says the same thing, more quietly:

next((user for user in users if user.is_admin), None)

Stream(users).filter(lambda user: user.is_admin).first()

None of these are faster than the itertools version — they compile to the same generators. They are shorter to read and, more usefully, they read in the order the work happens.

Laziness

Nothing executes until a terminal op runs. Intermediate ops just build a pipeline:

stream = Stream.from_iterable(count()).map(expensive)  # expensive() has not been called
stream.take(3).to_list()  # now it runs -- exactly 3 times

This is what makes infinite sources safe:

Stream.iterate(1, lambda n: n * 2).take_while(lambda n: n < 100).to_list()
# [1, 2, 4, 8, 16, 32, 64]

Single use

A stream is consumed once, like the iterator underneath it. Reusing one raises rather than silently yielding nothing:

stream = Stream.of(1, 2, 3)
stream.to_list()  # [1, 2, 3]
stream.to_list()  # StreamConsumedError

API

Creating a stream

Intermediate ops — lazy, return a new Stream

Terminal ops — consume the stream, return a value

Pipe operator

Any function taking a Stream can act as a stage, so you can extend the library without subclassing:

def errors_only(stream):
    return stream.filter(lambda line: line.startswith("ERROR"))


Stream(lines) | errors_only | (lambda s: s.take(10)) | list

Files

Stream.from_file streams a file's lines lazily and owns the handle:

with Stream.from_file("app.log") as lines:
    errors = lines.map(str.rstrip).filter(lambda line: line.startswith("ERROR")).to_list()

The file is closed on every exit path: exhausting the stream, stopping early with a break or a partial read, an exception escaping iteration, or leaving the with block. with remains the clearest way to express the intent, and it closes the file even if you never iterate at all.

Lines keep their trailing newline, matching open(). Pass encoding, errors or newline through as needed.

Handles you already have

Stream(handle) leaves the handle to whoever opened it — the right default, but it leaks if nobody follows up. Stream.from_handle makes the hand-off explicit:

Stream.from_handle(sock.makefile())  # the stream closes it
Stream.from_handle(sys.stdin, close=False)  # you keep it

With close=True (the default) the resource is closed on the same paths from_file covers; with close=False nothing is closed for you. It works for anything iterable with a close() — socket files, os.popen pipes, io.StringIO, database cursors — and raises TypeError for an iterable that has none. Streamlet never auto-detects handles in Stream(...): silently closing something you opened would break the whoever-opens-closes convention.

Async

AsyncStream[T] is the same API over an async source, driven with async for and await:

import asyncio

from streamlet import AsyncStream


async def fetch(url: str) -> str: ...


async def main() -> None:
    titles = await (
        AsyncStream.from_iterable(urls)
        .filter(lambda u: u.startswith("https://"))
        .map(fetch)
        .take(10)
        .to_list()
    )

Every op that takes a function accepts either a plain function or a coroutine function, so .map(str) and .map(fetch) both work and both type correctly.

Concurrency

.map(fetch) awaits one item at a time. .map_concurrent(fetch) runs several at once:

results = await AsyncStream.from_iterable(urls).map_concurrent(fetch, limit=8).to_list()

At most limit calls are in flight, and items are pulled from the source only as slots free up — so it stays lazy and works on infinite sources. Results come back in source order by default; pass ordered=False to get each one as soon as it is ready.

If the mapper raises, the original exception propagates (not an ExceptionGroup) and every in-flight call is cancelled. Abandoning the stream early — a take, a break — cancels them too.

A runnable fan-out demo lives in examples/concurrent_fan_out.py:

uv run python examples/concurrent_fan_out.py

sequential : 24 requests in 1.95s (1 at a time)
concurrent : 24 requests in 0.24s (up to 8 at once) -- 8.0x faster
first infra user: user-02 -- fetched 8/24 users, not all

Sources can be sync or async: AsyncStream.of(...), .from_iterable(list), .from_async_iterable(agen), .empty(), .iterate(seed, fn), .generate(fn), and .concat(...) — which mixes both kinds.

Terminal ops are coroutines, so they need await. Everything else matches Stream: the same intermediate ops, the same single-use semantics, the same laziness.

Cleanup

Each stage closes the one behind it, so a pipeline that stops early finalises its source right away — a bounded take, a short-circuiting first/any/all, or an exception all unwind the whole chain. That matters when the source holds something: a connection, a cursor, a handle.

One case Python gives no hook for is a bare break out of an async for, which leaves the outermost stage suspended. Wrap the iterator when that matters:

from contextlib import aclosing

async with aclosing(stream.map(fetch).__aiter__()) as items:
    async for item in items:
        if done(item):
            break

Typing

Stream[T] is generic and ships a py.typed marker, so element types flow through a chain:

Stream.of(1, 2, 3)          # Stream[int]
  .map(str)                 # Stream[str]
  .to_list()                # list[str]

Some ops are restricted by self-type and fail at type-check time, not runtime:

• .sum() only on numeric streams
• .join() only on Stream[str]
• .to_set(), .distinct(), .group_by() require hashable items

Development

uv sync                    # install project + dev tools (needs Python 3.11+)
uv run pytest              # tests
uv run ruff check .        # lint
uv run mypy                # type check (strict)

Run everything the way CI does:

uv run ruff check . && uv run ruff format --check . && uv run mypy && uv run pytest

Versioning

Streamlet follows semantic versioning. Before 1.0.0 a breaking change bumps the minor version and everything else bumps the patch version. Changes are recorded in CHANGELOG.md.

Releases are cut from a tag: pushing vX.Y.Z builds the artifacts and checks the tag against the version in pyproject.toml. Publishing the GitHub release for that tag uploads them to PyPI through trusted publishing, so no token lives in this repository.

Roadmap

• Core Stream[T], intermediate + terminal ops, laziness tests
• __or__ pipe chaining, constructors, group_by, edge cases
• Stream.from_file — context-managed resource streams
• AsyncStream — bounded concurrent map
• Docstrings, README examples
• PyPI release

License

MIT