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.

Shared executor budget (2.4.0)

BoundedExecutor wraps a caller-owned executor. Its limit counts unfinished calls across every query using that wrapper, including calls already running. Each query's buffer still bounds its retained futures and results. Closing the wrapper leaves the underlying pool open.

fpstreams.BoundedExecutor

Bases: Executor

Limit unfinished submissions across callers sharing one borrowed executor.

Submitters wait in arrival order when the budget is full. Completed or cancelled futures release their slots; retained results remain subject to each flow's buffer. Submit through the same wrapper to share its budget. Tasks submitted directly to the underlying executor are outside that budget.

Shutdown closes this wrapper and optionally cancels or waits for its own work. It never shuts down the underlying executor. As with a thread pool, callbacks must not wait for nested submissions that require their own occupied slots.

Parameters:

Name	Type	Description	Default
executor	Executor	Caller-owned thread, process, or other concurrent executor.	required
max_pending	int	Maximum number of unfinished submitted calls, including running calls.	required

Raises:

Type	Description
TypeError	If executor is not an Executor or max_pending is not an integer.
ValueError	If max_pending is less than one.

_executor instance-attribute

_executor = executor

_max_pending instance-attribute

_max_pending = limit

_condition instance-attribute

_condition = Condition()

_waiters instance-attribute

_waiters: deque[object] = deque()

_futures instance-attribute

_futures: set[Future[Any]] = set()

_admitting instance-attribute

_admitting = 0

_shutdown instance-attribute

_shutdown = False

_cancel_on_shutdown instance-attribute

_cancel_on_shutdown = False

max_pending property

max_pending: int

Return the shared capacity, including running calls.

pending property

pending: int

Return the current number of unfinished or being-submitted calls.

__init__

__init__(executor: Executor, *, max_pending: int) -> None

_reserve

_reserve() -> None

submit

submit(
    fn: Callable[P, T], /, *args: args, **kwargs: kwargs
) -> Future[T]

Submit a call, blocking for shared capacity without changing its Future.

_release

_release(future: Future[Any]) -> None

shutdown

shutdown(
    wait: bool = True, *, cancel_futures: bool = False
) -> None

Close admissions and clean up this wrapper's work, keeping the pool open.

File partitions (2.4.0)

FileCheckpoint computes one output file per stable input partition and processing version. See the recovery recipe for file ownership, retries, and storage requirements.

fpstreams.FileCheckpoint

Reuse completed file partitions after verifying their input and output bytes.

A callback receives the input path and a temporary output path. It must close one regular output file before returning. A successful run publishes that file and JSON metadata together by renaming their directory on the same filesystem. Callback and validation failures publish nothing. Cleanup errors can occur after publication; a retry verifies and reuses the completed entry. Publication guarantees atomic visibility, not durability after power loss. The directory must be trusted local storage.

Inputs must remain stable during execution. Use a new version whenever code, options, or external dependencies can change the result. Only the callback's file output is checkpointed; external side effects need their own idempotency. Concurrent runs of different partitions are independent. A duplicate running partition raises BlockingIOError instead of repeating its callback.

Parameters:

Name	Type	Description	Default
directory	str | PathLike[str]	Directory in which completed partitions and persistent lock files live.	required
version	str	Nonempty processing-version token covering callback logic and options.	required

_directory instance-attribute

_directory = Path(directory).resolve()

_version instance-attribute

_version = version

__init__

__init__(
    directory: str | PathLike[str], *, version: str
) -> None

_completed

_completed(entry: Path, expected: dict[str, Any]) -> Path

Reject incomplete or modified entries without replaying their callbacks.

run

run(
    source: str | PathLike[str],
    function: Callable[[Path, Path], object],
) -> Path

Return a verified output path, computing and publishing it only on a miss.

Parameters:

Name	Type	Description	Default
source	str | PathLike[str]	Stable regular input file; its resolved path and SHA256 identify a partition.	required
function	Callable[[Path, Path], object]	Receives input and temporary output paths; its return value is ignored.	required

Returns:

Type	Description
Path	The completed output file, ready to pass to another file-based stage.

Raises:

Type	Description
BlockingIOError	If this partition is already being computed.
ValueError	If input changes, output is missing, or a completed entry is corrupt.
OSError	If input or checkpoint storage cannot be read or written.

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.

_NO_USER_SELECTION_TRACKING module-attribute

_NO_USER_SELECTION_TRACKING = object()

_user_selection_error module-attribute

_user_selection_error: ContextVar[object] = ContextVar(
    "fpstreams_user_selection_error",
    default=_NO_USER_SELECTION_TRACKING,
)

_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.

_SelectionTracking

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.

_record_user_selection_error

_record_user_selection_error(error: SelectionError) -> None

Record a user failure only within an active drop-null evaluation.

_record_missing_selection_error

_record_missing_selection_error(
    error: SelectionError, origin: object
) -> None

_cleanup_error_summary

_cleanup_error_summary(
    error: BaseException, *, prefix: str = "cleanup failed:"
) -> str

Describe cleanup failures even when their exception message cannot be formatted.