snakestream 1.0.0

Snakestream is a Java 8 Stream-style API for Python, built from the ground up on async/await. Chain map, filter, sorted, flat_map, distinct and friends into a lazy pipeline, hand every one of them a sync or an async function interchangeably, and await a terminal operation to run it.

names = await (
    Stream(user_ids)
    .parallel()
    .map(fetch_user)  # async def — awaited for you, concurrently
    .filter(lambda u: u.active)
    .map(lambda u: u.name)  # plain def — same chain, no ceremony
    .collect(to_list())
)

Install

pip install snakestream     # or: uv add snakestream

Python 3.14+.

Quick start

import asyncio
from snakestream import Stream
from snakestream.collectors import to_list


async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.05)  # pretend this is a network call
    return {"id": user_id, "name": f"user-{user_id}", "active": user_id != 3}


async def main() -> None:
    names = await (
        Stream([1, 2, 3, 4, 5])
        .parallel()
        .map(fetch_user)
        .filter(lambda u: u["active"])
        .map(lambda u: u["name"])
        .collect(to_list())
    )
    print(names)  # ['user-1', 'user-2', 'user-4', 'user-5']


asyncio.run(main())

The five fetches overlap, and the result still comes back in encounter order.

Why Snakestream?

Async and sync callables are interchangeable, everywhere. Every user-supplied function — predicate, mapper, comparator, consumer, collector part, close handler — may be def or async def. You never wrap, never gather, never think about it. Awaitability is classified once per callable, not once per element.

Nothing runs until you ask for a value. Intermediate operations queue work and return a new stream; a terminal operation drives it. Sources stay lazy all the way through, so an infinite generator with a .limit(10) downstream pulls ten elements.

stream = Stream(count()).map(expensive).filter(is_valid)  # nothing has run yet
first_ten = await stream.limit(10).collect(to_list())  # exactly ten pulls

.parallel() is one word and it applies to the whole pipeline. Contiguous batches of elements are dispatched onto their own OS threads, each batch racing its own elements concurrently. I/O-bound pipelines speed up on any interpreter; CPU-bound ones speed up for real on the free-threaded build (3.14t). Switch back with .sequential(). See About .parallel().

It breaks apart nested comprehensions. A fluent chain reads top-to-bottom, and adding a step in the middle is one line rather than a re-nesting.

by_team = await Stream(users).collect(grouping_by(lambda u: u.team, counting()))
roster = await Stream(users).sorted(comparing(lambda u: u.name)).map(str).collect(joining(", "))

The API is Java's, deliberately. If you know Stream, Collectors and Comparator from Java 8, you already know this library — same method names, same semantics, same ordering guarantees. The API tables below are total over Java 8's surface: every method has a row saying implemented, skipped-and-why, or not-yet.

It is pure Python with zero dependencies.

Features

• Create a stream from a List, Generator, AsyncGenerator, Iterator, AsyncIterator or just an object
• Process your stream with both synchronous and asynchronous functions
• Switch between parallel and sequential mode (real parallelism on the free-threaded build)
• Encounter order preserved under .parallel(), or opt out with unordered() for more throughput
• ~20 collectors and a composable comparator, both mirroring Java
• Autoclose streams with with, async with or contextlib
• Pythonic protocols on top of the Java surface: async for, with, a + b
• Generate indefinite streams simpler than in Java

Scope

This is a Python streaming API that brings over the feature set Java 8 introduced with its streams API. Once we reach some sort of feature parity with Java 8, maybe we move on to the improvements in Java 9. There will never be complete parity, because the languages differ: we don't really speak about arrays in Python, we use lists and sets; and where Java streams lean on functional interfaces, Python is already a functional language, so Suppliers, Consumers and all of that are just regular functions here. So that's the road map as of now — we get as close as we can with a reasonable effort put into it.

Building a stream from a source

Stream(source) is the normalizing constructor and the idiomatic way to build a stream from something you already have: a List, Generator, AsyncGenerator, Iterator, AsyncIterator, or a bare object. It spreads any of those into one element per item — Stream([1, 2, 3]) is a stream of three elements — while dict, str, bytes, bytearray and memoryview values, and anything with neither __iter__ nor __next__, are treated as a single scalar element instead. It is the same constructor every Stream(...) call in this README uses, called out here because it has no counterpart in the parity tables that follow.

That absence is deliberate rather than an oversight. The tables are total over Stream, BaseStream, Collectors and Comparator, and Stream(source) is none of those — its closest Java counterparts, Collection.stream() and Arrays.stream(T[]), are methods on types this library does not have. Stream.of(*args: T) in the table below is the true parity row: it matches Java's of(T...) exactly, treating every argument as one element regardless of arity. Spreading a single iterable's items - what Stream.of() did through 0.3.5 - is Stream(source), not Stream.of(source).

The generate() function

Java's Stream.generate(supplier) is omitted here, because Python already has generators — pass one straight to Stream(...) and it becomes the source:

from itertools import count

first_ten_squares = await Stream(count()).map(lambda n: n * n).limit(10).collect(to_list())

About .parallel()

Unlike Java's parallelStream(), snakestream's .parallel() does not use a process pool — there is no pickling boundary to cross, since a Spliterator decomposes the stream's own composed chain into batches that run in-process. But it does now run on separate OS threads: a .parallel() pipeline dispatches contiguous batches of elements via asyncio.to_thread, each batch running its own copy of the chain on its own thread, rather than racing asyncio tasks cooperatively over a shared generator on one thread the way it used to.

Whether that buys CPU-bound work a real speedup depends on the interpreter build. On the ordinary GIL-enabled build, only one thread runs Python bytecode at a time, so CPU-bound work sees the same result as before — no measurable speedup, confirmed at ~1.0x — aside from the usual GIL release during I/O and certain C-extension calls. On the free-threaded build (3.14t, PEP 779) there is no GIL, so batches genuinely run Python code in parallel and CPU-bound work benefits for real — measured at ~2x on a large-enough source, and small sources benefit too: a 200-element source measures ~2.0x-2.1x and an 800-element one ~1.2x-1.3x, since the batch-size ramp no longer drains a small source's remainder into a single worker's second batch the way a one-step jump to the steady-state batch size used to — see benchmark-findings.md in openspec/changes/spread-small-sources-across-workers for the figures and parallel-worker-utilisation (openspec/specs/) for the distribution guarantee itself. I/O-bound work (e.g. a mapper that awaits a network call) benefits on both builds and at any size, since it always did — measured at up to 54x in the same benchmark. See benchmark-findings.md in openspec/changes/archive/2026-09-04-fork-join-executor-and-spliterator for the numbers behind all of this.

We know .concurrent()/CONCURRENCY would be the more idiomatic name for what this used to do, but keeping .parallel()/WORKERS turned out to be the right call for a different reason than originally planned — the naming didn't need to survive a future switch to multiprocessing, it needed to survive a switch to threads.

Where the switch is written is Java's rule exactly: the mode belongs to the pipeline rather than to a stage, so .map(f).parallel() and .parallel().map(f) build the same thing and the last switch before the terminal governs the whole chain. What diverges is the receiver. Java's parallel() writes the flag onto the pipeline's source stage and returns this, so the reference you called it on is the returned stream and keeps working under the new mode:

Stream<T> s = list.stream();
Stream<T> p = s.parallel();   // s == p, and s still usable

Here it returns a new stream and consumes the receiver, so the second line leaves s raising IllegalStateException. That is the same derive-and-consume rule map(), filter() and every other intermediate operation follow, applied to mode switches as well — Java can return this only because the flag is mutable state on a stage, and this library keeps no mutable per-stream state to flip. Assign the result and use that:

s = Stream([1, 2, 3]).parallel()
count = await s.map(fetch).count()

Auto Close

Stream is Java's AutoCloseable and Python's context manager at once, so with on the stream directly runs whatever you registered with on_close():

with Stream(rows) as stream:
    letters = await stream.map(to_letter).distinct().collect(to_list())

contextlib.closing() works too, and is what older examples use — it only needs a .close() attribute, which Stream has:

from contextlib import closing

with closing(Stream(rows)) as stream:
    letters = await stream.map(to_letter).distinct().collect(to_list())

This is especially useful when you subclass Stream to wrap something IO-like that holds a resource needing release after the stream: put that logic behind on_close() and the with handles the rest.

A close handler may be sync or async — on_close() accepts either. close() stays synchronous: it runs sync handlers and refuses one whose result is awaitable, raising StreamBuildException and pointing you at aclose()/async with. aclose() is the asynchronous twin — it awaits an awaitable handler's result and runs a sync handler exactly as close() does, one at a time, in registration order:

stream = Stream(rows).on_close(conn.aclose)

async with stream as s:
    it = await s.map(parse).collect(to_list())

contextlib.aclosing() works the same way contextlib.closing() does, for the same reason — Stream implements __aenter__/__aexit__ beside __enter__/__exit__. Use with/closing() when every handler is sync; reach for async with/aclosing() only once a handler needs awaiting.

Your subclass's __init__ runs once per pipeline — at your own MyStream(...) call, not again for each .map() or .parallel() — so a resource you acquire there is acquired once and shared by every stage, and the one close() releases it once. Your __init__ may also take whatever arguments you like; nothing requires it to mirror Stream.__init__:

class DsnStream(Stream):
    def __init__(self, dsn):
        self.conn = connect(dsn)
        super().__init__(self.conn.rows())
        self.on_close(self.conn.close)


with closing(DsnStream("db://x")) as stream:
    rows = await stream.map(parse).filter(is_valid).collect(to_list())

Python's data model

The parity tables below are total over Java 8's surface, and Python's dunder methods are not Java methods — so they live here rather than becoming rows nobody wrote.

Three of these are parity rather than expansion: Java's stream satisfies its own language's iteration, resource and toString protocols, and these are Python's equivalents.

API

The three tables below are total over Java 8's surface: every method of Stream, BaseStream, Collectors and Comparator has a row, whether or not it exists here. A Java 8 method with no row is a defect in the table, not a silence to interpret. The leftmost column has three states:

The tables cover Stream and BaseStream only. IntStream, LongStream and DoubleStream are skipped wholesale, for the reason map_to_int()'s row below gives: they exist in Java to avoid autoboxing primitives and to expose numeric-only operations a generic Stream<T> cannot offer, and Python numbers are already objects with no boxing cost. Every primitive-specialization skip row in the tables is a consequence of that one decision rather than an independent judgement.

StreamSupport is skipped too, but only six of its eight statics fall to that same reason — the intStream/longStream/doubleStream pairs. The two generic stream(Spliterator, boolean) overloads do not, and they are the ones that turn a Spliterator back into a stream. spliterator() exists here and that return trip does not, which is a real gap rather than a primitive-specialization consequence; it is tracked in roadmap/ as spliterator-round-trip rather than settled here.

Stream

Collectors

Collector lives in snakestream.collector; every factory in the table below lives in snakestream.collectors, the same split Java draws between the Collector interface and the Collectors class that holds the factories.

Collector(supplier, accumulator, combiner=None, finisher=None, characteristics=frozenset()) is the type every factory below returns, mirroring Java's Collector<T,A,R>: supplier() creates a fresh accumulation container, accumulator(container, element) mutates it per element (sync or async; its return value is ignored), and finisher(container) converts the finished container into the result, or the container itself is the result if finisher is omitted. combiner(container, container) merges two partial containers into one, left-biased in batch order, and is invoked under .parallel() wherever the collector supplies one and the source spans more than one batch (parallel-reduction); a collector with no combiner is never partitioned, and folds into a single container exactly as every collector did before this. characteristics is a Characteristics set - data, not a callable, so it is neither invoked nor awaited - defaulting to empty, so every existing Collector(...) call is unaffected. A Collector instance holds no other per-collection state, so the instance one of these factories returns is safe to reuse across streams and across concurrent collections. You can construct one directly for a custom reduction: Stream([1, 2, 3]).collect(Collector(list, lambda c, e: c.append(e))).

characteristics is a declaration a collector makes about itself, not an instruction any operation performs. collect() is its only reader: under parallel() it reads UNORDERED to decide whether the collector is owed a reorder barrier. UNORDERED promises that any two orderings of the same elements collect to an equal result - == on the result's own type - and promises nothing about the iteration order of that result.

Comparator

comparing() lives in snakestream.comparator, alongside the comparator semantics sorted(), min(), max(), min_by() and max_by() already share.

Migration

Every breaking change is recorded in CHANGELOG.md, newest first, with what changed, whether the break is loud or silent, and what to do about it. Until release 1.0.0, expect that list to grow.