PyStreamAPI

PyStreamAPI is a Python stream library inspired by the Java Stream API, adding Pythonic features for clean, declarative, and efficient data processing. It supports both sequential and parallel streams with lazy evaluation.

from pystreamapi import Stream

Stream.of([" ", '3', None, "2", 1, ""]) \
    .filter(lambda x: x is not None) \
    .map(str) \
    .map(lambda x: x.strip()) \
    .filter(lambda x: len(x) > 0) \
    .map(int) \
    .sorted() \
    .for_each(print) # Output: 1 2 3

Installation

pip install streams.py

from pystreamapi import Stream

🎉 PyStreamAPI is now ready to process your data

Why PyStreamAPI?

• Sequential and parallel streams out of the box
• Lazy execution for efficient processing of large datasets
• 100% test coverage
• Pythonic API — clean, readable, and expressive
• 111+ built-in conditions for filtering and matching
• Declarative error handling with configurable error levels
• Built-in loaders for CSV, JSON, XML, YAML and TOML files

Building a Stream

PyStreamAPI provides two stream types — Stream (general-purpose) and NumericStream (for numerical data with statistics) — each available in sequential and parallel flavors.

Stream.of([1, 2, 3])               # auto-selects sequential or numeric
Stream.parallel_of([1, 2, 3])      # parallel stream
Stream.sequential_of([1, 2, 3])    # sequential stream
Stream.of_noneable(None)           # returns empty stream when source is None
Stream.iterate(0, lambda n: n + 2) # infinite stream (use .limit())
Stream.concat(Stream.of([1, 2]), Stream.of([3, 4]))  # merge streams

For the full API reference see the docs.

Conditions

Over 111 ready-to-use conditions across strings, numbers, types, and dates — combine them freely with one_of():

from pystreamapi import Stream
from pystreamapi.conditions import prime, even, one_of

Stream.of([1, 2, 3, 4, 5]) \
    .filter(one_of(even(), prime())) \
    .for_each(print) # 2, 3, 4, 5

Error Handling

Control error behavior per-operation with error_level():

from pystreamapi import Stream, ErrorLevel

Stream.of([" ", '3', None, "2", 1, ""]) \
    .error_level(ErrorLevel.IGNORE) \
    .map_to_int() \
    .sorted() \
    .for_each(print) # Output: 1 2 3

Available levels: RAISE (default), IGNORE, WARN. See the error handling docs for details.

Data Loaders

Load data from files directly into a stream — no manual parsing needed:

Loader	Extra required	Description
csv	—	CSV files with optional type casting and delimiter
json	[json_loader]	JSON files or strings (streaming via ijson)
xml	[xml_loader]	XML files or strings with node path access
yaml	—	YAML files or strings
toml	—	TOML files or strings

from pystreamapi import Stream
from pystreamapi.loaders import csv

Stream.of(csv("data.csv", delimiter=";")) \
    .map(lambda x: x.name) \
    .for_each(print)

Install all optional extras at once: pip install 'streams.py[all]'

See the data loaders docs for full usage.

Documentation

Full documentation: pystreamapi.pickwicksoft.org

Bug Reports

Bug reports can be submitted in GitHub's issue tracker.

Contributing

Contributions are welcome! Please submit a pull request or open an issue.