loom.streaming.compiler

Streaming compiler: validates StreamFlow and produces CompiledPlan.

loom.streaming.compiler.compile_flow(flow, *, config)

Compile a flow into an immutable plan.

Parameters:

• flow (StreamFlow[Any, Any]) – User-declared streaming flow.

• config (ConfigContext | Mapping[str, Any]) – Canonical runtime config context used for binding resolution and Kafka settings extraction.

Returns:

Immutable CompiledPlan ready for adapter wiring.

Raises:

CompilationError – If binding resolution or any validation phase fails.

Return type:

CompiledPlan

exception loom.streaming.compiler.CompilationError(issues)

Bases: Exception

Raised when a StreamFlow fails validation or cannot be compiled.

Aggregates every CompilationIssue found in a compilation run. Bare strings are accepted for backwards compatibility and normalized to issues with code StreamingErrorCode.UNSPECIFIED.

issues

Structured issues, one per failure, each carrying a machine-readable code, component, and optional config field.

Parameters:

issues (Sequence[CompilationIssue | str]) – Issues (or legacy message strings) collected by the compiler.

Return type:

None

property errors: list[str]

Human-readable messages, one per issue (legacy accessor).

class loom.streaming.compiler.CompilationIssue(*, code, message, component='', field=None)

Bases: LoomFrozenStruct

One structured compiler failure.

Parameters:

• code (StreamingErrorCode) – Machine-readable StreamingErrorCode.

• message (str) – Human-readable description; the aggregated exception text is built from these messages.

• component (str) – Node or config section the issue points at (for example "router branch 'vip'").

• field (str | None) – Optional config field involved (for example "kafka.consumer.group_id").

prefixed(prefix)

Return a copy scoped under a branch prefix.

Used when branch validation nests issues under fork branch X / router branch X / broadcast branch N contexts. The message keeps the historical "{prefix}: {message}" format.

Parameters:

prefix (str)

Return type:

CompilationIssue

class loom.streaming.compiler.CompiledMongoCDCSource(settings, collections, watch_options, shape)

Bases: object

Resolved MongoDB CDC source configuration.

Parameters:

• settings (MongoSourceConfig)

• collections (tuple[str, ...])

• watch_options (Mapping[str, object])

• shape (StreamShape)

class loom.streaming.compiler.CompiledMultiSource(settings, topics, dispatch, shape, decode_strategy)

Bases: object

Resolved heterogeneous Kafka source with a pre-built dispatch table.

Parameters:

• settings (ConsumerSettings) – Resolved Kafka consumer settings.

• topics (tuple[str, ...]) – Kafka topic names to subscribe to.

• dispatch (DispatchTable) – Pre-built table mapping message_type and error payload_type strings to their concrete Python types, plus explicit wire-error payload types.

• shape (StreamShape) – Declared source shape.

• decode_strategy (Literal['record', 'batch']) – Whether to decode records individually or in batches.

class loom.streaming.compiler.CompiledNode(node, input_shape, output_shape, path=())

Bases: object

DSL node annotated with input/output shapes.

Parameters:

• node (object)

• input_shape (StreamShape)

• output_shape (StreamShape)

• path (tuple[int, ...])

class loom.streaming.compiler.CompiledPlan(name, source, nodes, output, error_routes, needs_async_bridge, terminal_sinks=<factory>, terminal_storage_sinks=<factory>)

Bases: object

Immutable result of compiling a StreamFlow.

Parameters:

• name (str)

• source (CompiledSingleSource | CompiledMultiSource | CompiledMongoCDCSource)

• nodes (tuple[CompiledNode, ...])

• output (CompiledSink | None)

• error_routes (dict[ErrorKind, CompiledSink])

• needs_async_bridge (bool)

• terminal_sinks (dict[tuple[int, ...], CompiledSink])

• terminal_storage_sinks (dict[tuple[int, ...], CompiledStorageSink])

class loom.streaming.compiler.CompiledSingleSource(settings, topics, payload_type, shape, decode_strategy)

Bases: object

Resolved Kafka source with a single payload type and decode strategy.

Parameters:

• settings (ConsumerSettings)

• topics (tuple[str, ...])

• payload_type (type[LoomStruct | LoomFrozenStruct])

• shape (StreamShape)

• decode_strategy (Literal['record', 'batch'])

class loom.streaming.compiler.CompiledSink(settings, topic, partition_policy, dlq_topic=None)

Bases: object

Resolved Kafka sink with partitioning.

Parameters:

• settings (ProducerSettings)

• topic (str)

• partition_policy (PartitionPolicy[LoomStruct | LoomFrozenStruct] | None)

• dlq_topic (str | None)

class loom.streaming.compiler.CompiledStorageSink(node, config, database_config=None)

Bases: object

Resolved storage sink with its DSL node and pre-fetched config section.

The adapter calls node.build_partition(config, worker_index, worker_count, bridge) at startup once per Bytewax worker to obtain the SinkPartition that handles epoch writes and shutdown. Storage backends that require async execution receive the adapter-managed AsyncBridge.

Parameters:

• node (Any) – The IntoSink node as declared in the DSL.

• config (SqlAlchemySinkConfig | DeltaSinkConfig | ClickHouseSinkConfig) – Resolved storage sink config for the selected backend.

• database_config (SqlAlchemyDatabaseConfig | None) – Resolved shared SQLAlchemy database config, if needed.

class loom.streaming.compiler.StreamingErrorCode(value)

Bases: StrEnum

Enumeration of all streaming compiler failure codes.

loom.streaming.compiler.walk_process_nodes(nodes)

Yield every DSL node reachable from nodes, recursing into branches.

Public traversal helper over the process tree: recurses into Router, Fork, Broadcast, ExpandRoutes and scoped With/WithAsync processes. Used by the compiler phases and by runtime guards that must inspect nested nodes.

Parameters:

nodes (Iterable[object])

Return type:

Iterable[object]