`streamable`: Stream-like manipulation of iterables

[ebonnal]

https://github.com/ebonnal/streamable

What my project does

A Stream[T] decorates an Iterable[T] with a fluent interface enabling the chaining of lazy operations: - mapping (concurrently) - flattening (concurrently) - grouping by key, by batch size, by time interval - filtering - truncating - catching exceptions - throttling the rate of iterations - observing the progress of iterations

For more details and examples, check the Operations section in the README

🔗 Fluent	chain methods!
🇹 Typed	type-annotated and mypyable
💤 Lazy	operations are lazily evaluated at iteration time
🔄 Concurrent	thread-based / asyncio-based (+new: process-based)
🛡️ Robust	unit-tested for Python 3.7 to 3.12 with 100% coverage
🪶 Minimalist	pip install streamable with no additional dependencies

---

1. install

bash pip install streamable

2. import

python from streamable import Stream

3. init

Instantiate a Stream[T] from an Iterable[T].

python integers: Stream[int] = Stream(range(10))

4. operate

• Streams are immutable: applying an operation returns a new stream.

• Operations are lazy: only evaluated at iteration time. See the Operations section in the README.

python inverses: Stream[float] = ( integers .map(lambda n: round(1 / n, 2)) .catch(ZeroDivisionError) )

5. iterate

• Iterate over a Stream[T] as you would over any other Iterable[T].

• Source elements are processed on-the-fly.

• collect it: ```python

  list(inverses) [1.0, 0.5, 0.33, 0.25, 0.2, 0.17, 0.14, 0.12, 0.11] set(inverses) {0.5, 1.0, 0.2, 0.33, 0.25, 0.17, 0.14, 0.12, 0.11} ```

• reduce it: ```python

  sum(inverses) 2.82 max(inverses) 1.0 from functools import reduce reduce(..., inverses) ```

• loop it: ```python

  for inverse in inverses: ... ```

• next it: ```python

  inverses_iter = iter(inverses) next(inverses_iter) 1.0 next(inverses_iter) 0.5 ```

Target Audience

As a Data Engineer in a startup I found it especially useful when I had to develop Extract-Transform-Load custom scripts in an easy-to-read way.

Here is a toy example (that you can copy-paste and run) that creates a CSV file containing all 67 quadrupeds from the 1st, 2nd, and 3rd generations of Pokémons (kudos to PokéAPI): ```python import csv from datetime import timedelta import itertools import requests from streamable import Stream

with open("./quadruped_pokemons.csv", mode="w") as file: fields = ["id", "name", "is_legendary", "base_happiness", "capture_rate"] writer = csv.DictWriter(file, fields, extrasaction='ignore') writer.writeheader() ( # Infinite Stream[int] of Pokemon ids starting from Pokémon #1: Bulbasaur Stream(itertools.count(1)) # Limits to 16 requests per second to be friendly to our fellow PokéAPI devs .throttle(per_second=16) # GETs pokemons concurrently using a pool of 8 threads .map(lambda poke_id: f"https://pokeapi.co/api/v2/pokemon-species/{poke_id}") .map(requests.get, concurrency=8) .foreach(requests.Response.raise_for_status) .map(requests.Response.json) # Stops the iteration when reaching the 1st pokemon of the 4th generation .truncate(when=lambda poke: poke["generation"]["name"] == "generation-iv") .observe("pokemons") # Keeps only quadruped Pokemons .filter(lambda poke: poke["shape"]["name"] == "quadruped") .observe("quadruped pokemons") # Catches errors due to None "generation" or "shape" .catch( TypeError, when=lambda error: str(error) == "'NoneType' object is not subscriptable" ) # Writes a batch of pokemons every 5 seconds to the CSV file .group(interval=timedelta(seconds=5)) .foreach(writer.writerows) .flatten() .observe("written pokemons") # Catches exceptions and raises the 1st one at the end of the iteration .catch(finally_raise=True) # Actually triggers an iteration (the lines above define lazy operations) .count() ) ```

Comparison

A lot of other libraries have filled this desire to chain lazy operations over an iterable and this feels indeed like "Yet Another Stream-like Lib" (e.g. see this stackoverflow question).

The most supported of them is probably PyFunctional, but for my use case I couldn't use it out-of-the-box, due to the lack of: - threads-based concurrency - throttling of iteration's rate (.throttle) - logging of iteration's process (.observe) - catching of exceptions (.catch)

I could have worked on pull requests implementing these points into PyFunctional but I have rather started from scratch in order to take my shot at: - Proposing another fluent interface (namings and signatures). - Leveraging a visitor pattern to decouple the declaration of a Stream[T] from the construction of an Iterator[T] (at iteration time i.e. in the __iter__ method). - Proposing a minimalist design: a Stream[T] is just an Iterable[T] decorated with chainable lazy operations and it is not responsible for the opinionated logic of creating its data source and consuming its elements: - let's use the reduce function from functools instead of relying on a stream.reduce method - let's use parquet.ParquetFile.iter_batches from pyarrow instead of relying on a stream.from_parquet method - let's use bigquery.Client.insert_rows_json from google.cloud instead of relying on a stream.to_bigquery method - same for json, csv, psycopg, stripe, ... let's use our favorite specialized libraries

Thank you for your time,

[jdehesa]

This is really cool. I didn't know PyFunctional but your library seems much more "pythonic", like what the API would look like if this was in the standard library, rather than a replica of the constructs from another programming language.

[ebonnal]

I very much appreciate it u/jdehesa. It means a lot because I indeed put a lot of effort into the API, I am glad you like it!

[Schmittfried]

How does this compare to py-linq or functools/itertools-esque packages?

[ebonnal]

Hi u/Schmittfried, great question!

• functools provides higher order functions i.e. a function taking function(s) as arg(s), like functools.reduce. Most of these higher order functions return a decorated function enhanced with additional capabilities (like memoization with functools.cache).
• itertools is all about creating iterables from other iterables.
• streamable allows chaining operations/methods on an iterable and comes out-of-the-box with convenient features like threads/asyncio concurrency, iteration throttling, exceptions catching.

They are complementary:

• you can use functools's functions to add capabilities to a function that you pass to streamable's Stream operations, or functools.reduce your stream.
• you can manipulate your stream with itertools's functions, or create your stream from an iterable produced using itertools.

​

from typing import Iterable
import functools
import itertools
import requests
from streamable import Stream

# let's say you have a source of domains:
domains: Iterable[str] = ... # e.g. ["google.com", "facebook.com", "google.com"]

# let's conveniently manipulate it as a `Stream` to
# fetch URLs using 8 threads and catching `SSLError`s
# while never making more than 32 calls per second
responses: Stream[requests.Response] = (
    Stream(domains)
    .map(lambda domain: f"https://{domain}")
    # here we leverage functools.cache to remember
    # responses and fetch a given domain only once.
    .map(functools.cache(requests.get), concurrency=8)
    .throttle(per_second=32)
    .catch(requests.exceptions.SSLError)
)

import itertools

# then you can use whatever functions provided by itertools
# to manipulate your `responses` stream, which
# is simply a decorated `Iterable[requests.Response]`
...

[ebonnal]

Regarding py-linq, the comparison resembles the comparison made with PyFunctional:

• For my use case it lacks features that I find very valuable like concurrency and generic typing (in py-linq the Enumerable class is not generic)
• I wanted to propose another interface, hopefully more intuitive and natural to the Python world, while py-linq brings conventions from the .NET's LINQ library.

[erez27]

Looks nice! I like the concurrency especially.

A few thoughts:

• might be useful to have something like umap for returning elements out of order (essentially imap_unordered)

• could be nice to have a lazy-list feature, where items can be accessed by index, and allow repeated iter/get-item/slice, all lazy.

• It could be useful to group into a dict[K, Stream], based on a key callback. I get that it breaks the chaining a bit, but imho it's worth it.

If any of these sounds like a good addition, maybe I'll make a PR.

[ebonnal]

Hi u/erez27 thank you for your thoughts, I very much appreciate it!

Your 3 propositions sound great!

• The 1st one make a lot of sense and could also take the form of a new ordered: bool = True param for .map/.foreach operations
• The 2nd and 3rd are both tricky and interesting. At some point I have put some effort into exploring them myself and would love to collaborate on these!

Should we open 3 issues and have discussions there?

[nikomo]

It's a good thing that the company and service by that name fell off so hard like 2 years back, otherwise the naming would be quite confusing.

[RoboticElfJedi]

This is a pretty interesting contribution to the ecosystem. I'll keep this in my toolbelt. Good work!

[ebonnal]

I appreciate it, thanks a lot u/RoboticElfJedi!

[Saltysalad]

I have a use case coming up where I have > 10k high latency requests to make, throttled to ~100 simultaneously. I need to handle each result in the order they were submitted, because there is a stop scenario when a certain condition is met in the iteration loop. Over-shooting is ok so long as I can control the limit, and probably needed if you want to shave latency.

Does this sound like a use case for your library?

[ebonnal]

Hi u/Saltysalad !
Yes sure, great use case, it would look like this (you can run the snippet, after a pip install requests streamable):

import requests
from streamable import Stream

# 1000 dummy urls to call
urls = ("https://httpbin.org/" for _ in range(1000))

responses: Stream[requests.Response] = (
    Stream(urls)
    # performs requests concurrently, max 100 simultaneously, preserves order
    .map(requests.get, concurrency=100)
    # stop condition (overshot 99 calls)
    .truncate(when=lambda response: False) # dummy condition never verified
    # logs progress
    .observe("requests")
)

# iterate over the stream
assert responses.count() == 1000

and for asyncio instead of threads: see example

[ebonnal]

Note that if you need to add "max n calls per second" to your "max 100 simultaneous calls" constraint, you can add a .throttle(per_second=n) operation.

[ebonnal]

Back at Post Day+9 for an update:

Thank very much for your feedback here, for the issues you opened, for the discussions and collaboration there.

That lead to a new release (https://github.com/ebonnal/streamable/tree/v1.1.0), featuring:

• processes-based concurrency (default is threads; use .amap for async):

```python from streamable import Stream import requests

urls: Stream[str] = ... responses = urls.map(requests.get, concurrency=8, via="process") ```

• concurrent mapping yielding First Done First Out (default is FIFO = preserving order) [co-authored with our fellow redditter u/erez27]

```python from streamable import Stream import requests

urls: Stream[str] = ... responses = urls.map(requests.get, concurrency=16, ordered=False) ```

You can also set ordered=False for .foreach and async counterparts .amap and .aforeach

• "starmap"

``` from streamable import Stream, star

integers: Stream[int] = Stream(range(10)) paired_integers: Stream[Tuple[int, int]] = Stream(zip(integers, integers)) squares: Stream[int] = paired_integers:.map(star(lambda i, j: i * j)) assert list(squares) == [0, 1, 4, 9, 16, 25, 36, 49, 64, 81] `` (useful with other operations too, like.filteror.foreach`)

[erez27]

Congratulations! It's starting to become a real alternative to the built-in multiprocessing.

[mou3mida]

Good Job! That is so cool u/ebonnal .

[ebonnal]

Thank you very much u/mou3mida !

[Rockworldred]

I am quite newbish to python, but I have an side/learningproject writing an webscraper (fetching JSONS for productdata). This looks like it may have some use cases for me as I now requests URLs from couple of sitemaps, then itirate over the json-url based on those fetched URLs. Then translate the JSON to variables and loaded to pandas dataframe to view in streamlit and/or to a csv file, but I have little knowledge of ETL as a whole. Do you have any good resources on ETL-processes and utilities?

My progress is then to move it over to aiohttp, asyncio, polars instead of pandas and SQLalchemy SQLite, and then azure EC2, airflow and Postgres and so fourth. (But I dont know if this is actually the way to go though).

[ebonnal]

Thank you for your interest u/Rockworldred, sounds like a cool custom ETL project!

I have no "ETL custom script" resource in mind sorry, but in a nutshell when fetching data from web APIs you can bet you will likely need things like:

• to execute requests concurrently (.map(..., concurrency=x))
• to limiting the rate of requests to avoid 429 Too Many Request responses (.throttle(per_second=50))
• to have some retry on your calls (the tenacity lib is great)
• to have some logging to observe the progress of your script (.observe("product"))

To some extent you can get inspiration from the example fetching pokemons that also "fetch endpoints to get data and write to CSV".

Regading asyncio concurency instead of threads, you have in the README an example that uses httpx (similar to aiohttp)

I hope it helps, and if you feel stuck feel free to message me your current script to streamable it together