tinystream 0.0.7

Yet another python streams library

tinystream / python-streams

This is a simple and lightweight Streams API inspired by Java Streams with support for type hinting.

This package is release as tinystream at pypi.

Basic API

from tinystream import Stream

stream = Stream.of([1, 2, 3, 4, 5]) # of_many(*), of_dict()

stream \
    .map(lambda x: x + 1) \       # flatmap(), peek(), map_key()
    .filter(lambda x: x > 2) \    # filter_key()
    .sorted(int, reverse=True) \  # sort()
    .reverse() \                  # collect(), count()
    .limit(2) \
    .concat([4]) \
    .sum()                        # reduce(), max(), min()

Typehinting

Since Python does not support typed lambdas, this library implements a workaround.

from tinystream import Stream

stream = Stream.of(["A", "B", "C"], str)

This is not necessary when typing is used:

from tinystream import Stream
from typing import List

list: List[str] = ["A", "B", "C"]
stream = Stream.of(list)

Type hinting the given type:

from dataclasses import dataclass

@dataclass
class Node:
    name: str
    parent: "Node" = None

for lambdas:

parent = Node(name="B")
child = Node(name="A", parent=parent)

stream = Stream.of([child])
assert next(stream.map(lambda x: x.parent, typehint=Node)).name == "B"

This is not necessary when you pass a mapping function:

def map_parent(n: Node):
    return n.parent

assert next(stream.map(map_parent)).name == "B"

Typed dictionaries

Dictionaries are streamed as tuple(key, value)

children = {"a": Node(name="Child")} 
stream = Stream.of_dict(children)
for item in stream:
    # item[0] is known as str
    # item[1] is known as Node

This is the same like (but without known types):

stream = Stream.of(children)

End of stream

Calling methods like sum(), collect(), count()... will end the stream.

More features

Filter by existing key

items_with_name = Stream.of([child]).filter_key("name")

Map object name attribute

names = Stream.of([child]).map_key("name")

Deep mapping of name attributes

list = [
   {"node": Node(name="Node A")},
   {"node": Node(name="Node B")},
   {"node": Node(name="Node C")},
   {"node": Node(name="Node D")},
]
Stream.of(list).map_keys(("node", "name"))

Collected join

all_names = Stream.of([child]).map_key("name").join(", ")

Comparison with other libraries

There are a couple of other implementation to fulfill similar requirements.

• https://github.com/alemazzo/Python-Java-Stream
• https://github.com/ramsteak/streams
• https://pypi.org/project/fluentpy
• https://github.com/JaviOverflow/python-streams (outdated)
• https://github.com/9seconds/streams/ (outdated)
• https://github.com/tolsac/streampy (outdated)
• Apache Spark

Run the tests

PYTHONPATH="." pytest --cov=tinystream -n 4 tests/

Release update

1. Update version in setup.py
2. Package library

   python setup.py sdist
   

3. Publish library

   twine upload dist/tinystream-[version].tar.gz
   

References

• https://github.com/MichaelKim0407/tutorial-pip-package
• https://packaging.python.org/en/latest/guides/making-a-pypi-friendly-readme/