Gatherers

A Gatherer is a stateful intermediate operation. Unlike a Collector, it can emit zero or more values while the rest of the pipeline is still running.

Use the regular constructor for the iterable-returning interface:

from fpstreams import Gatherer, flow

pairs = Gatherer(
    initializer=lambda: None,
    integrator=lambda _state, value: (value, -value),
)

assert flow([1, 2]).gather(pairs).to_list() == [1, -1, 2, -2]

Gatherer.of() uses a push interface. Its integrator receives the current state, the input item, and a Downstream. Return False to stop reading the source. Gatherer.of_sequential() declares that its state cannot be combined in parallel. and_then() composes two gatherers without collecting the intermediate output.

Gatherer

fpstreams.Gatherer dataclass

Bases: Generic[T, S, R]

Describe a stateful intermediate operation in legacy iterable or push-callback form.

of classmethod

of(
    initializer_or_integrator: PushIntegrator[None, T, R],
    *,
    finisher: PushFinisher[None, R] | None = None,
    combiner: Combiner[None] | None = None,
    greedy: bool = False,
) -> Gatherer[T, None, R]

of(
    initializer_or_integrator: Callable[[], S],
    integrator: PushIntegrator[S, T, R],
    *,
    finisher: PushFinisher[S, R] | None = None,
    combiner: Combiner[S] | None = None,
    greedy: bool = False,
) -> Gatherer[T, S, R]

of(
    initializer_or_integrator: Callable[..., Any],
    integrator: Callable[..., Any] | None = None,
    *,
    finisher: Callable[..., Any] | None = None,
    combiner: Callable[..., Any] | None = None,
    greedy: bool = False,
) -> Gatherer[Any, Any, Any]

Build a push-mode gatherer in stateless or explicitly initialized form.

Omitting integrator makes the first callable a stateless integrator with None state. The finisher defaults to no output. combiner and greedy are retained as metadata for state composition; sequential integration still stops on rejection.

of_sequential classmethod

of_sequential(
    initializer_or_integrator: PushIntegrator[None, T, R],
    *,
    finisher: PushFinisher[None, R] | None = None,
    greedy: bool = False,
) -> Gatherer[T, None, R]

of_sequential(
    initializer_or_integrator: Callable[[], S],
    integrator: PushIntegrator[S, T, R],
    *,
    finisher: PushFinisher[S, R] | None = None,
    greedy: bool = False,
) -> Gatherer[T, S, R]

of_sequential(
    initializer_or_integrator: Callable[..., Any],
    integrator: Callable[..., Any] | None = None,
    *,
    finisher: Callable[..., Any] | None = None,
    greedy: bool = False,
) -> Gatherer[Any, Any, Any]

Build the push-mode form through of while always storing no state combiner.

and_then

and_then(
    other: Gatherer[R, Any, RR],
) -> Gatherer[T, Any, RR]

Feed this gatherer directly into other through a short-circuiting bridge.

Both states and proceed flags are retained together. This finisher sends left-finisher output through the right integrator before running the right finisher. A composite state combiner exists only when both component gatherers provide one.

Downstream

Downstream.push(value) emits a value and reports whether downstream still wants more. Check is_rejecting() before expensive work when downstream may have short-circuited.

fpstreams.Downstream

Bases: Generic[R]

Forward gatherer output while making downstream rejection permanent.

push

push(value: R) -> bool

Offer one value to the callback and latch rejection when it returns False.

Once rejecting, the channel returns False without invoking the callback. Callback results must be actual booleans so short-circuit behavior cannot depend on truthiness.

is_rejecting

is_rejecting() -> bool

Return the latched downstream rejection state.