loom.streaming.bytewax

Bytewax runtime adapter for Loom streaming flows.

This package requires bytewax to be installed.

Usage:

from loom.streaming.bytewax import build_dataflow

flow = build_dataflow(compiled_plan)

exception loom.streaming.bytewax.RuntimeConfigurationError(issues)

Bases: Exception

Raised when a valid compiled plan cannot run under this runtime config.

Distinct from CompilationError on purpose. The plan itself is sound — it compiled — and nothing is being compiled when this is raised. What fails is the combination of that plan with the requested runtime, which is only knowable while the dataflow is assembled. Reporting it as a compilation failure would send the reader looking for a mistake in their flow definition instead of in their runtime configuration.

issues

Structured issues, each carrying a machine-readable code, the component involved, and the configuration field to change.

Parameters:

issues (Sequence[CompilationIssue]) – Issues describing why the runtime rejected the plan.

Return type:

None

loom.streaming.bytewax.build_dataflow(plan, *, observability_runtime=None, source=None, sink=None, terminal_sinks=None, error_sinks=None)

Build a Bytewax Dataflow from a compiled plan.

Parameters:

• plan (CompiledPlan)

• observability_runtime (ObservabilityRuntime | None)

• source (Any | None)

• sink (Any | None)

• terminal_sinks (Mapping[tuple[int, ...], Any] | None)

• error_sinks (Mapping[ErrorKind, Any] | None)

Return type:

Any

loom.streaming.bytewax.build_dataflow_with_shutdown(plan, *, observability_runtime=None, source=None, sink=None, terminal_sinks=None, error_sinks=None, bridge=None, commit_tracker=None)

Build a Bytewax Dataflow and expose its shutdown callback.

Parameters:

• plan (CompiledPlan) – Compiled flow plan.

• observability_runtime (ObservabilityRuntime | None) – Optional observability runtime for lifecycle events.

• source (Any | None) – Optional Bytewax source override (used in tests).

• sink (Any | None) – Optional Bytewax sink override (used in tests).

• terminal_sinks (Mapping[tuple[int, ...], Any] | None) – Optional per-branch sink overrides.

• error_sinks (Mapping[ErrorKind, Any] | None) – Optional per-kind error sink overrides.

• bridge (AsyncBridge | None) – Pre-configured AsyncBridge. When None, a default asyncio bridge is created if the plan requires async execution. Pass an explicit bridge to control backend and uvloop settings.

• commit_tracker (CommitCompletionPort | None)

Return type:

_BuiltDataflow

class loom.streaming.bytewax.BytewaxRecoverySettings(db_dir, backup_interval_ms=None)

Bases: LoomFrozenStruct

Recovery settings for the Bytewax runtime.

Parameters:

• db_dir (str) – Local recovery directory shared by this execution.

• backup_interval_ms (int | None) – Optional snapshot backup interval in milliseconds.

class loom.streaming.bytewax.BytewaxRuntimeConfig(workers_per_process=1, process_id=None, addresses=None, epoch_interval_ms=None, recovery=None, async_backend='asyncio', use_uvloop=False, force_shutdown_timeout_ms=None)

Bases: LoomFrozenStruct

Runtime settings for executing a Bytewax dataflow.

Delivery guarantee: By default Kafka offsets are committed by the consumer’s auto-commit policy and are not tied to downstream success. When enable.auto.commit is disabled in the consumer settings, Loom tracks item completion through the adapter and commits offsets only after the downstream branch finishes successfully.

Parameters:

• workers_per_process (int) – Number of worker threads in this process.

• process_id (int | None) – Optional cluster process id.

• addresses (tuple[str, ...] | None) – Optional cluster address list, including this process.

• epoch_interval_ms (int | None) – Optional epoch duration in milliseconds.

• recovery (BytewaxRecoverySettings | None) – Optional recovery settings.

• async_backend (Literal['asyncio', 'trio']) – Anyio backend used for AsyncBridge when the flow contains WithAsync nodes. Accepted values: "asyncio" (default) or "trio".

• use_uvloop (bool) – Replace the asyncio event loop with uvloop for lower latency async I/O. Only effective when async_backend is "asyncio" and the platform is not Windows.

• force_shutdown_timeout_ms (int | None) – Maximum milliseconds to wait for in-flight async tasks to finish during shutdown. When exceeded, the portal thread is abandoned and a warning is logged. None waits indefinitely (default).

exception loom.streaming.bytewax.DuplicateErrorSinkError

Bases: LoomStreamingError

Raised when two registered sinks cover the same ErrorKind.

class loom.streaming.bytewax.KafkaPartitionedSource(source, commit_tracker=None, observability=None)

Bases: FixedPartitionedSource[KafkaRecord[bytes], int | None]

One Bytewax input partition per Kafka partition.

Partition keys are "{topic}:{index}" — a durable contract: they are also the Bytewax recovery-state keys, so the format must stay stable (: is not a legal topic character; parse with rpartition).

Consumers are created lazily per partition inside build_part() and pinned with assign — no subscribe, no group membership: the consumer group acts purely as an offset store. Start-offset precedence:

resume_state (Bytewax recovery) > committed group offset > auto_offset_reset

Parameters:

• source (_CompiledKafkaSource)

• commit_tracker (KafkaCommitTracker | None)

• observability (ObservabilityRuntime | None)

list_parts()

List one partition key per Kafka partition of every source topic.

Note: with auto.create.topics.enable=true on the broker, asking for a missing topic may auto-create it (confluent metadata caveat).

Return type:

list[str]

build_part(step_id, for_part, resume_state)

Build one partition: lazy consumer, assign, floor, tracker binding.

Parameters:

• step_id (str)

• for_part (str)

• resume_state (int | None)

Return type:

_KafkaSourcePartition

bind_commit_tracker(commit_tracker)

Bind or clear the source commit tracker (adapter duck-typed hook).

Parameters:

commit_tracker (KafkaCommitTracker | None)

Return type:

None

class loom.streaming.bytewax.RegisteredSink(*args, **kwargs)

Bases: Protocol

Protocol a registrable sink class must satisfy.

sink_type

Unique type identifier. Must match the type field in the YAML config.

Type:

ClassVar[str]

config_type

msgspec Struct type used to deserialize the YAML config section.

Type:

ClassVar[type]

Example:

class ClickHouseErrorTableSink:
    sink_type = "clickhouse_error_table"
    config_type = ClickHouseErrorTableSinkConfig

    @classmethod
    def build_binding(cls, cfg, ctx):
        ...
        return RuntimeSinkBinding(purpose="errors", sink=..., kinds=(ErrorKind.TASK,))

class loom.streaming.bytewax.RuntimeSinkBinding(purpose, sink, kinds=())

Bases: LoomFrozenStruct

Resolved sink binding returned by a RegisteredSink.

Parameters:

• purpose (Literal['errors', 'terminal', 'audit']) – Role of this sink within the runner.

• sink (object) – Instantiated sink object ready for use.

• kinds (tuple[ErrorKind, ...]) – For purpose="errors", the ErrorKind values this sink handles.

class loom.streaming.bytewax.StreamingRunner(*, _registry=None)

Bases: object

Wire a StreamFlow declaration into the real Bytewax runtime.

Parameters:

_registry (SinkRegistry | None) – Optional pre-built SinkRegistry for testing purposes.

classmethod from_context(flow, *, config, observability_runtime=None)

Build a runner from a resolved config context.

Parameters:

• flow (StreamFlow[Any, Any])

• config (ConfigContext)

• observability_runtime (ObservabilityRuntime | None)

Return type:

StreamingRunner

classmethod from_yaml(flow, path, *, resolvers=(), observability_runtime=None)

Load config from YAML and build a runner.

Parameters:

• flow (StreamFlow[Any, Any]) – Flow to compile.

• path (str) – Local path or cloud URI of the YAML config file.

• resolvers (Sequence[ConfigResolver]) – Resolvers for ${name:key} placeholders, registered before the built-in secrets and ssm defaults. A resolver named like a default replaces it.

• observability_runtime (ObservabilityRuntime | None) – Optional runtime override.

Return type:

StreamingRunner

classmethod from_dict(flow, config, *, observability_runtime=None)

Build a runner from a plain Python config mapping.

Parameters:

• flow (StreamFlow[Any, Any])

• config (dict[str, Any])

• observability_runtime (ObservabilityRuntime | None)

Return type:

StreamingRunner

register_sink(sink_class)

Register a sink class to be resolved from YAML config at run time.

Parameters:

sink_class (type) – Class that satisfies the RegisteredSink protocol.

Raises:

TypeError – If sink_class does not implement RegisteredSink.

Return type:

None

build_dataflow()

Assemble the Bytewax Dataflow and keep shutdown state.

Return type:

Dataflow

prepare_run(*, error_sinks=None, runtime=None)

Prepare one executable dataflow and its shutdown callback.

Releases any resources from a previous prepare_run() call before building the new dataflow, so calling this method twice is safe.

Parameters:

• error_sinks (Mapping[ErrorKind, ErrorSink] | None) – Optional mapping of ErrorKind to ErrorSink instances. When provided, these sinks override the default error routing built from the compiled plan.

• runtime (BytewaxRuntimeConfig | None) – Optional runtime override; defaults to the runner’s config-loaded runtime settings.

Returns:

A bundle containing the assembled Dataflow and a shutdown callable that releases adapter-owned resources.

Return type:

_PreparedStreamingRun

run(*, flow=None, config=None, config_path=None, error_sinks=None, runtime=None, resolvers=())

Execute the compiled dataflow with the real Bytewax runtime.

Emits Scope.POLL_CYCLE around the full invocation (including preparation) and Scope.FLOW around the blocking cli_main call. Starts the Prometheus scrape server before execution when configured.

Parameters:

• flow (StreamFlow[Any, Any] | None) – Optional flow override. When provided together with config or config_path, the runner is reconfigured before execution.

• config (ConfigContext | None) – Optional config context. Used together with flow to reconfigure the runner.

• config_path (str | Path | None) – Optional path to a YAML config file. Loaded and used like config when provided.

• error_sinks (Mapping[ErrorKind, Any] | None) – Optional explicit error sinks. When provided, these take precedence over registry-resolved sinks.

• runtime (BytewaxRuntimeConfig | None) – Optional runtime override. When omitted, uses the runner’s config-loaded runtime settings.

• resolvers (Sequence[ConfigResolver]) – Resolvers for ${name:key} placeholders in config_path, registered before the built-in secrets and ssm defaults. A resolver named like a default replaces it. Only meaningful together with config_path.

Raises:

ValueError – If resolvers is given together with config instead of config_path; config is already resolved and resolvers would never be applied.

Return type:

None

shutdown()

Release adapter resources after a run or failed build.

Return type:

None