Skip to content

Errors and runtime values

The root-level pipeline errors listed below inherit from FlowError. Exceptions raised by user callbacks, Python protocols, parsers, optional libraries, the filesystem, and database drivers keep their original type unless an adapter documents a more specific boundary. Some public subpackages define additional errors for their own storage or runtime contracts.

Error hierarchy

Error Raised when
FlowError Base class for fpstreams operation errors
FlowConsumedError A one-shot source is evaluated after it has already been consumed
EmptyFlowError An element-requiring terminal such as first() receives no item
SelectionError A field, index, path, or expression cannot select a value
DuplicateKeyError An output field, JSON object key, pivot cell, or join suffix would be ambiguous
NativeUnsupportedError A plan forced to native cannot be represented by the native engine
BufferLimitError A configured record, buffer, partition, fan-out, or output budget is exceeded

Catch the narrow error that your application can recover from. Catching FlowError is useful at a pipeline boundary, but it does not include arbitrary callback or third-party-library exceptions.

from fpstreams import EmptyFlowError, flow

try:
    value = flow([]).first()
except EmptyFlowError:
    value = None

When an optional adapter is missing, install the extra named in its ImportError message. A missing optional integration is not converted to FlowError.

SpillLimits

SpillLimits sets hard limits for partitioned joins and grouping. These cover partition size, matches per key, output rows, and repartition depth.

fpstreams.SpillLimits dataclass

Bound partition rows and bytes, join fan-out, output rows, and repartition depth.

Field Default Meaning
max_partition_rows 100,000 Maximum rows retained in one partition
max_partition_bytes 64 MiB Maximum estimated bytes in one partition
max_matches_per_key 100,000 Maximum join fan-out for one key
max_output_rows 1,000,000 Maximum total records emitted by the bounded operation
max_repartition_depth 3 Maximum recursive repartition attempts

Limits must be positive integers except repartition depth, which may be zero. Exceeding a limit raises BufferLimitError and triggers query cleanup.

SummaryStatistics

flow(values).collect(Collectors.summarizing()) returns one mutable statistics value containing count, sum, minimum, maximum, and a derived average.

fpstreams.SummaryStatistics dataclass

Mutable count, numeric sum, minimum, maximum, and derived average state.

count class-attribute instance-attribute 

count: int = 0

sum class-attribute instance-attribute 

sum: float = 0.0

min class-attribute instance-attribute 

min: float = float('inf')

max class-attribute instance-attribute 

max: float = float('-inf')

average property 

average: float

Divide the running sum by count, returning 0.0 for empty state.

Returns:

Type Description
float

sum / count when at least one value was accepted, otherwise 0.0.

accept 

accept(value: float) -> None

Increment count and update sum, minimum, and maximum in place.

Parameters:

Name Type Description Default
value float

Numeric value supporting addition and ordering with prior values.

 required

__init__ 

__init__(
    count: int = 0,
    sum: float = 0.0,
    min: float = float("inf"),
    max: float = float("-inf"),
) -> None

Empty input is normalized to count and sum zero, minimum and maximum 0.0, and average 0.0. Flow.describe() is a separate convenience terminal that returns a dictionary and may include sample standard deviation for numeric input.

Reducer laws

Parallel or tree reduction can change grouping. fpstreams does not infer that an arbitrary function is associative, commutative, or has a valid identity. Reducer metadata makes those requirements explicit.

fpstreams.Reducer

Bases: Collector[T, S, R], Generic[T, S, R]

A collector with a mandatory state merger and validated algebraic laws.

The inherited initializer, step, finish, early-completion predicate, and native metadata define sequential collection. combine and laws additionally authorize partitioned state reduction.

initializer instance-attribute 

initializer: Callable[[], S]

step instance-attribute 

step: Callable[[S, T], S]

finish class-attribute instance-attribute 

finish: Callable[[S], R] = _identity

combine class-attribute instance-attribute 

combine: Callable[[S, S], S] | None = None

done class-attribute instance-attribute 

done: Callable[[S], bool] = _never_done

native class-attribute instance-attribute 

native: Any | None = None

_lifecycle_revision class-attribute instance-attribute 

_lifecycle_revision: int = field(
    init=False, default=0, repr=False, compare=False
)

__slots__ class-attribute instance-attribute 

__slots__ = ('laws',)

laws instance-attribute 

laws: ReducerLaws

__post_init__ 

__post_init__() -> None

Require callable lifecycle hooks and an optional callable state combiner.

__call__ 

__call__(values: Iterable[T]) -> R

Initialize, step, and finish over one traversal with deterministic close.

The completion predicate is evaluated once for each newly produced state. Closing in finally preserves the public early-stop and failure behavior for generator sources without importing the higher-level program module.

__init__ 

__init__(
    initializer: Callable[[], S],
    step: Callable[[S, T], S],
    finish: Callable[[S], R] = _identity,
    *,
    merge: Callable[[S, S], S],
    laws: ReducerLaws,
    done: Callable[[S], bool] = _never_done,
    native: Any | None = None,
) -> None

Initialize a reducer after validating its merge callable and law declaration.

reduce 

reduce(values: Iterable[T]) -> R

Run the inherited collector state machine over values and finish its state.

fpstreams.ReducerLaws dataclass

Declare merge laws, order requirements, empty behavior, and state growth.

Reducers require a true associative merge and a true identity. Commutative reducers cannot also be marked order-sensitive. state tells planners whether partial state is constant-sized or grows with input, while provenance records who established the laws.

associative instance-attribute 

associative: Literal[True]

commutative instance-attribute 

commutative: bool

order_sensitive instance-attribute 

order_sensitive: bool

identity instance-attribute 

identity: Literal[True]

empty_input instance-attribute 

empty_input: EmptyInputPolicy

state instance-attribute 

state: StateProfile

provenance instance-attribute 

provenance: LawProvenance

__post_init__ 

__post_init__() -> None

Validate mandatory laws, compatible flags, and enum/profile field types.

to_dict 

to_dict() -> dict[str, Any]

Serialize law flags and nested state metadata to plain Python values.

__init__ 

__init__(
    associative: Literal[True],
    commutative: bool,
    order_sensitive: bool,
    identity: Literal[True],
    empty_input: EmptyInputPolicy,
    state: StateProfile,
    provenance: LawProvenance,
) -> None

fpstreams.ReductionExplanation dataclass

Summarize whether a collector has trusted laws and a declared state combiner.

mergeable instance-attribute 

mergeable: bool

combine_declared instance-attribute 

combine_declared: bool

laws instance-attribute 

laws: ReducerLaws | None

to_dict 

to_dict() -> dict[str, Any]

Serialize mergeability, combiner presence, and optional reducer laws.

__init__ 

__init__(
    mergeable: bool,
    combine_declared: bool,
    laws: ReducerLaws | None,
) -> None

ReducerLawError reports a missing or contradicted law. LawProvenance records whether a law was declared, derived, or verified according to the reducer API.

Root error module

Exceptions for flow consumption, selection, collection, and execution failures.

_CANONICAL_SELECTION_ERROR module-attribute 

_CANONICAL_SELECTION_ERROR = SelectionError

FlowError

Bases: Exception

Base class for errors raised by fpstreams operations.

FlowConsumedError

Bases: FlowError

Raised when code tries to evaluate an already-consumed one-shot flow.

EmptyFlowError

Bases: FlowError

Raised when an empty flow cannot satisfy an element-requiring terminal.

SelectionError

Bases: FlowError, LookupError

Raised when a field, index, path, or expression selector cannot resolve a value.

DuplicateKeyError

Bases: FlowError, ValueError

Raised when dictionary collection encounters a key with no overwrite policy.

NativeUnsupportedError

Bases: FlowError

Raised when a plan forced to the native engine contains an unsupported operation.

BufferLimitError

Bases: FlowError

Raised when a bounded record or buffer exceeds its configured byte or item limit.