drava_common.config¶

One place that reads and validates a Drava pipeline.yaml.

Previously the same schema was parsed three different ways (yaml-cpp in the runtime, a hand-rolled indent parser in publisher_util.py, and another hand-rolled parser in benchmark_two_stages.py). This module is the single Python reader: it uses PyYAML when available and degrades to a tiny built-in scalar parser otherwise, so examples do not hard-depend on PyYAML.

The runtime itself (C++) remains authoritative and reads the same YAML via yaml-cpp; this module intentionally mirrors that schema and never invents new runtime behavior.

Exceptions¶

PipelineConfigError

Raised when a pipeline.yaml is missing, malformed, or mis-wired.

Classes¶

`StageConfig`	One `stages:` entry from a `pipeline.yaml`.
`PipelineConfig`	A parsed, normalized `pipeline.yaml`.

Functions¶

`load_yaml`(→ dict)	Load a YAML mapping from `path`.
`load_pipeline_config`(→ PipelineConfig)	Load and normalize a pipeline.yaml into a `PipelineConfig`.
`validate_pipeline`(→ list[str])	Check that a pipeline is internally consistent and return warnings.

Module Contents¶

exception drava_common.config.PipelineConfigError¶

Bases: Exception

Raised when a pipeline.yaml is missing, malformed, or mis-wired.

drava_common.config.load_yaml(path: pathlib.Path | str) → dict¶

Load a YAML mapping from path.

Uses PyYAML if installed; otherwise a small fallback parser that supports the subset of YAML the Drava schema uses (nested maps, - name: list items, scalar values, # comments). Always returns a dict.

class drava_common.config.StageConfig¶

One stages: entry from a pipeline.yaml.

The runtime, ingress, egress, and metrics sections are kept as raw dicts so unknown keys are preserved.

name: str¶

runtime: dict¶

ingress: dict¶

egress: dict¶

metrics: dict¶

property threads: int | None¶: Worker thread count (runtime.threads), or None if unset.

property callback_batch: int | None¶: Callback batch size (runtime.callback_batch), or None if unset.

class drava_common.config.PipelineConfig¶

A parsed, normalized pipeline.yaml.

Returned by load_pipeline_config(). raw holds the full parsed mapping; the other fields are convenience views over it.

path: pathlib.Path¶

raw: dict¶

name: str¶

transport_type: str¶

nats_url: str¶

stages: list[StageConfig]¶

stage(name: str) → StageConfig¶: Return the stage named name, or raise PipelineConfigError.

property stage_names: list[str]¶: Stage names in declaration order.

drava_common.config.load_pipeline_config(path: pathlib.Path | str | None = None) → PipelineConfig¶

Load and normalize a pipeline.yaml into a PipelineConfig.

If path is None, uses $DRAVA_STAGE_CONFIG (the same env var the runtime reads). Raises PipelineConfigError on missing/invalid config.

drava_common.config.validate_pipeline(cfg: PipelineConfig) → list[str]¶

Check that a pipeline is internally consistent and return warnings.

Errors (raise PipelineConfigError):

no stages;
duplicate stage names;
for NATS transport, a non-terminal stage whose egress stream/subject does not match the next stage’s ingress stream/subject (the classic “nothing is flowing” typo).

Warnings (returned, not raised):

a non-terminal stage with egress.forward_eos: false (downstream will never see end-of-stream);
a stage missing ingress stream/subject on NATS transport.