pyrio 1.1.0

Functional-style Streams library for processing collections and querying files (json, toml, yaml, xml, csv, tsv). Provides easy integration with itertools.

PYRIO

Functional-style Streams API library

Facilitates processing of collections and iterables using fluent APIs.
Gives access to files of various types (json, toml, yaml, xml, csv and tsv) for reading and executing complex queries
Provides easy integration with itertools
(NB: Commonly used itertools 'recipes' are included as part of the main APIs)

How to use

Creating streams

• stream from iterable

Stream([1, 2, 3])

• from variadic arguments

Stream.of(1, 2, 3)

• empty stream

Stream.empty()

• infinite ordered stream

Stream.iterate(0, lambda x: x + 1)

• infinite unordered stream

import random

Stream.generate(lambda: random.random())

• infinite stream with given value

Stream.constant(42)

• concat
  (concatenate several streams together or add new streams to the current one)

Stream.concat((1, 2, 3), [5, 6]).to_list()
Stream.of(1, 2, 3).concat([4, 5]).to_list()

• prepend
  (prepend iterable to current stream)

Stream([2, 3, 4]).prepend(0, 1).to_list()

---

Intermediate operations

• filter

Stream([1, 2, 3]).filter(lambda x: x % 2 == 0)

• map

Stream([1, 2, 3]).map(str).to_list()
Stream([1, 2, 3]).map(lambda x: x + 5).to_list()

• filter map
  (filters out all None or falsy values (if falsy=True) and applies mapper function to the elements of the stream)

Stream.of(None, "foo", "", "bar", 0, []).filter_map(str.upper, falsy=True).to_list()

["FOO", "BAR"]

• reduce
  (returns Optional)

Stream([1, 2, 3]).reduce(lambda acc, val: acc + val, identity=3).get()

---

Terminal operations

Collectors

• collecting result into list, tuple, set

Stream([1, 2, 3]).to_list()
Stream([1, 2, 3]).to_tuple()
Stream([1, 2, 3]).to_set()

• into dict

class Foo:
    def __init__(self, name, num):
        self.name = name
        self.num = num
        
Stream([Foo("fizz", 1), Foo("buzz", 2)]).to_dict(lambda x: (x.name, x.num))

{"fizz": 1, "buzz": 2}

In the case of a collision (duplicate keys) the 'merger' functions indicates which entry should be kept

collection = [Foo("fizz", 1), Foo("fizz", 2), Foo("buzz", 2)]
Stream(collection).to_dict(collector=lambda x: (x.name, x.num), merger=lambda old, new: old)

{"fizz": 1, "buzz": 2}

• alternative for working with collectors is using the collect method

Stream([1, 2, 3]).collect(tuple)
Stream.of(1, 2, 3).collect(list)
Stream.of(1, 1, 2, 2, 2, 3).collect(set)
Stream.of(1, 2, 3, 4).collect(dict, lambda x: (str(x), x * 10))

• grouping

Stream("AAAABBBCCD").group_by(collector=lambda key, grouper: (key, len(grouper)))

{"A": 4, "B": 3, "C": 2, "D": 1}

coll = [Foo("fizz", 1), Foo("fizz", 2), Foo("fizz", 3), Foo("buzz", 2), Foo("buzz", 3), Foo("buzz", 4), Foo("buzz", 5)]
Stream(coll).group_by(
    classifier=lambda obj: obj.name,
    collector=lambda key, grouper: (key, [(obj.name, obj.num) for obj in list(grouper)]))

{
  "fizz": [("fizz", 1), ("fizz", 2), ("fizz", 3)],
  "buzz": [("buzz", 2), ("buzz", 3), ("buzz", 4), ("buzz", 5)],
}

Other terminal operations

• for_each

Stream([1, 2, 3, 4]).for_each(lambda x: print(f"{'#' * x} ", end=""))

• count
  (returns the count of elements in the stream)

Stream([1, 2, 3, 4]).filter(lambda x: x % 2 == 0).count()

• sum

Stream.of(1, 2, 3, 4).sum() 

• find_first
  (searches for an element of the stream that satisfies a predicate, returns an Optional with the first found value, if any, or None)

Stream.of(1, 2, 3, 4).filter(lambda x: x % 2 == 0).find_first().get()

• find_any
  (searches for an element of the stream that satisfies a predicate, returns an Optional with some of the found values, if any, or None)

Stream.of(1, 2, 3, 4).filter(lambda x: x % 2 == 0).find_any().get()

• any_match
  (returns whether any elements of the stream match the given predicate)

Stream.of(1, 2, 3, 4).any_match(lambda x: x > 2)

• all_match
  (returns whether all elements of the stream match the given predicate)

Stream.of(1, 2, 3, 4).all_match(lambda x: x > 2)

• none_match
  (returns whether no elements of the stream match the given predicate)

Stream.of(1, 2, 3, 4).none_match(lambda x: x < 0)

• min
  (returns Optional with the minimum element of the stream)

Stream.of(2, 1, 3, 4).min().get()

• max
  (returns Optional with the maximum element of the stream)

Stream.of(2, 1, 3, 4).max().get()

• compare_with
  (compares linearly the contents of two streams based on a given comparator)

fizz = Foo("fizz", 1)
buzz = Foo("buzz", 2)
Stream([buzz, fizz]).compare_with(Stream([fizz, buzz]), lambda x, y: x.num == y.num)

• quantify
  (counts how many of the elements are Truthy or evaluate to True based on a given predicate)

Stream([2, 3, 4, 5, 6]).quantify(predicate=lambda x: x % 2 == 0)

---

Itertools integration

import itertools

Stream([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]).use(itertools.islice, start=3, stop=8)

Itertools 'recipes'

• tee

Stream([1, 2, 3]).ncycles(count=2).to_list()

---

Querying files with FileStream

• working with json, toml, yaml, xml files

FileStream("path/to/file").map(lambda x: f"{x.key}=>{x.value}").to_tuple()

(
  "abc=>xyz", 
  "qwerty=>42",
)

from operator import itemgetter

(FileStream("path/to/file")
    .filter(lambda x: "a" in x.key)
    .map(lambda x: (x.key, sum(x.value) * 10))
    .sorted(itemgetter(1), reverse=True)
    .map(lambda x: f"{str(x[1])}::{x[0]}")
    .to_list()) 

["230::xza", "110::abba", "30::a"]

FileStream reads data as series of Item objects with key/value attributes.

• querying csv and tsv files
  (each row is read as a dict with keys taken from the header row)

FileStream("path/to/file").map(lambda x: f"fizz: {x['fizz']}, buzz: {x['buzz']}").to_tuple() 

(
  "fizz: 42, buzz: 45",
  "fizz: aaa, buzz: bbb",
)

from operator import itemgetter

FileStream("path/to/file").map(itemgetter('fizz', 'buzz')).to_tuple()

(('42', '45'), ('aaa', 'bbb'))