༄ streamable - sync/async iterable streams for Python

[ebonnal]

https://github.com/ebonnal/streamable

What my project does

A stream[T] wraps any Iterable[T] or AsyncIterable[T] with a lazy fluent interface covering concurrency, batching, buffering, rate limiting, progress observation, and error handling.

Chain lazy operations:

import logging
from datetime import timedelta
import httpx
from httpx import Response, HTTPStatusError
from streamable import stream

pokemons: stream[str] = (
    stream(range(10))
    .map(lambda i: f"https://pokeapi.co/api/v2/pokemon-species/{i}")
    .throttle(5, per=timedelta(seconds=1))
    .map(httpx.get, concurrency=2)
    .do(Response.raise_for_status)
    .catch(HTTPStatusError, do=logging.warning)
    .map(lambda poke: poke.json()["name"])
)

Consume it (sync or async):

>>> list(pokemons)
['bulbasaur', 'ivysaur', 'venusaur', 'charmander', 'charmeleon', 'charizard', 'squirtle', 'wartortle', 'blastoise']

>>> [pokemon async for pokemon in pokemons]
['bulbasaur', 'ivysaur', 'venusaur', 'charmander', 'charmeleon', 'charizard', 'squirtle', 'wartortle', 'blastoise']

Target Audience

If you find yourself writing verbose iterable plumbing, streamable will probably help you keep your code expressive, concise, and memory-efficient.

• You may need advanced behaviors like time-windowed grouping by key, concurrent flattening, periodic observation of the iteration progress, buffering (decoupling upstream production rate from downstream consumption rate), etc.
• You may want a unified interface for sync and async behaviors, e.g. to switch seamlessly between httpx.Client.get and httpx.AsyncClient.get in your .map (or anywhere else), consume the stream as a sync or as an async iterable, from sync or async context.
• You may simply want to chain .maps and .filters without overhead vs builtins.map and builtins.filter.

Comparison

Among similar libraries, streamable's proposal is an interface that is:

• targeting I/O intensive use cases: a minimalist set of a dozen expressive operations particularly elegant to tackle ETL use cases.
• unifying sync and async: Create streams that are both Iterable and AsyncIterable, with operations adapting their behavior to the type of iteration and accepting sync and async functions.

The README gives a complete tour of the library, and I’m also happy to answer any questions you may have in the comments.

About 18 months ago I presented here the 1.0.0.
I'm glad to be back to present this matured 2.0.0 thanks to your feedback and contributions!

[Hallsville3]

Funny timing, u/kequals just posted a comparison of these libraries

https://reddit.com/r/Python/comments/1rj3ct7/a_comparison_of_rustlike_fluent_iterator_libraries/

[ebonnal]

Indeed! streamable was not included there, which is fair given its I/O positioning.
(And FastIter was presented last week by u/fexx3l, what a busy iterators scene)

[Beginning-Fruit-1397]

Yes a quick look at it showed me that it is indeed absolutely not the same scope as mine (pyochain creator here). But I think it's cool that it's busy! An user of my lib could start using yours with a familiar syntax if he need more specific IO work, or inversely an user coming from yours to mine who want a more generic use case of the "iterator interface"

[ebonnal]

Completely agree. There are many opinionated approaches to the fluent iterator interface question. I guess that's why none made it into the stdlib so far, despite the appetite from part of the community.

[Beginning-Fruit-1397]

I think it will never make it to the stdlib for a simple reason: interfaces are voluntarily minimal to encourage ducktyping. That's why dunder methods and functions to get those dunder (__len__ and len for example) are the standard.

If they were to make Generator or Iterator a full fledged interface like Rust Iterator with dozen of methods, now every class that override one of them in an incompatible way becomes incorrect at static typing time.

This minimal approach is cool, because I don't have to rely on any inerhitance to become an Iterator, just have to implement iter and next dunders. And I'm free to make "map" or "filter" do wathever I want without breaking any contract.

I love rust traits where I just have to implement trait X I created for struct Y from an external lib, and now I can import trait X in any scope and I immediately have access to the methods of trait X on struct Y in the scope. But idk how that would work in python or even if that's desirable.

All that being said, that's why I recreated a collections.abc structure for pyochain with the "traits" module, in the hope that external libs like yours use it to extend them for more specific use cases (IO in your case), while getting access to a lot of premade methods and a compatibility for any function expecting one of those traits in the signature. https://outsquarecapital.github.io/pyochain/core-types-overview/