GitHub

Streams

PreviousNext

Pull-based streams with backpressure, concurrency and sinks.

A Stream[A, E, R] emits any number of As, then ends or fails with E. Streams are pulled: nothing runs until a consumer asks for items, so a slow consumer slows the producer down (backpressure) and memory stays bounded.

basics.py
import pyfect as fx
from pyfect import stream

# A stream is a description too: nothing is pulled until it runs.
evens = stream.range(0, 100).filter(lambda n: n % 2 == 0).map(lambda n: n * n).take(4)

squares = fx.run_sync(evens.run_collect())
print(squares)
assert squares == [0, 4, 16, 36]


def lookup(n: int) -> fx.Effect[str]:
    return fx.sleep(0.01).as_(f"item {n}")


# map_effect runs an effect per item, here three at a time, keeping the order.
items = stream.from_iterable([1, 2, 3, 4]).map_effect(lookup, concurrency=3)
pairs = fx.run_sync(items.grouped(2).run_collect())
print(pairs)
assert pairs == [("item 1", "item 2"), ("item 3", "item 4")]

Build streams with stream.make, from_iterable, range, unfold, paginate, from_queue, from_async_iterable, tick, ... and transform them with methods: map, map_effect, filter, take, scan, grouped, zip, merge, buffer, debounce, throttle, retry, timeout. The ones that remove an error or a service (stream.catch_tag, stream.provide) are functions for .pipe, as for effects.

Sinks

A Sink consumes a stream and produces a result; run_collect, run_fold and run_for_each are shortcuts for common ones.

sinks.py
import pyfect as fx
from pyfect import sink, stream

numbers = stream.range(1, 11)

# A sink consumes a stream and produces one value.
assert fx.run_sync(numbers.run(sink.sum())) == 55
assert fx.run_sync(numbers.run(sink.take(3))) == [1, 2, 3]

# Streams of bytes or text: decode, split into lines, fold.
body = stream.from_iterable([b"alpha\nbe", b"ta\ngamma\n"])
lines = body.decode_text().split_lines()
assert fx.run_sync(lines.run_fold(lambda: 0, lambda count, _line: count + 1)) == 3

A pipeline

NDJSON records arriving in byte chunks, priced by a flaky service four at a time with retries, buffered so memory stays bounded, batched and written by a sink. examples/stream_pipeline.py in the repository is a longer version.

pipeline.py
"""NDJSON in, a flaky service called four at a time, batches out."""

from dataclasses import dataclass

import pyfect as fx
from pyfect import schedule, sink, stream
from pyfect.encoding import Json, ndjson


@dataclass
class Unavailable(fx.TaggedError):
    order_id: int


calls: dict[int, int] = {}


def price(order: Json) -> fx.Effect[int, Unavailable]:
    """Fails the first call for order 3, as a busy service would."""
    assert isinstance(order, dict) and isinstance(order["id"], int)
    order_id = order["id"]

    def call() -> fx.Effect[int, Unavailable]:
        calls[order_id] = calls.get(order_id, 0) + 1
        if order_id == 3 and calls[order_id] == 1:
            return fx.fail(Unavailable(order_id))
        return fx.sleep(0.001).as_(order_id * 100)

    # suspend: each retry calls the service again, instead of reusing one result
    return fx.suspend(call)


written: list[tuple[int, ...]] = []

# Byte chunks that split records anywhere, as an HTTP body would.
body = stream.from_iterable([b'{"id": 1}\n{"id"', b': 2}\n{"id": 3}\n{"id": 4}\n{"id": 5}\n'])

pipeline = (
    body.pipe_through_channel(ndjson.decode())
    .map_effect(
        lambda o: price(o).retry(schedule=schedule.exponential(0.001), times=3), concurrency=4
    )
    .buffer(capacity=8)  # pricing runs at most 8 ahead of the writer: bounded memory
    .grouped(2)
    .run(sink.for_each(lambda batch: fx.sync(lambda: written.append(batch))))
)

assert fx.run_sync_exit(pipeline) == fx.Success(None)
assert written == [(100, 200), (300, 400), (500,)]
assert calls[3] == 2  # retried once

Streams are pulled: nothing upstream runs until the sink asks for more. map_effect keeps at most four calls in flight and buffer lets pricing run at most eight items ahead of the writer, so a slow sink slows the whole pipeline instead of filling memory.

pullbyte chunksndjson.decode()map_effect(price,concurrency=4)buffer(capacity=8)grouped(2)sink.for_each(write)
pullbyte chunksndjson.decode()map_effect(price,concurrency=4)buffer(capacity=8)grouped(2)sink.for_each(write)

fx.encoding has framers for NDJSON, JSON arrays and Server-Sent Events.

Reference: pyfect.stream, pyfect.sink, pyfect.channel.