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(errors)

Bases: Exception

Raised when a StreamFlow fails validation or cannot be compiled.

Parameters:

errors (list[str]) – List of human-readable error messages.

Return type:

None

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.