symfonic.devtools.ownership.loader

loader

Load and schema-check the ownership manifest TOML (T1.2.7).

ManifestError

Bases: ValueError

The manifest file is structurally invalid.

load_manifest

load_manifest(path: Path) -> OwnershipManifest

Parse path into an :class:OwnershipManifest, or raise ManifestError.

Source code in src/symfonic/devtools/ownership/loader.py

def load_manifest(path: Path) -> OwnershipManifest:
    """Parse *path* into an :class:`OwnershipManifest`, or raise ManifestError."""
    with open(path, "rb") as handle:
        try:
            data = tomllib.load(handle)
        except tomllib.TOMLDecodeError as exc:
            raise ManifestError(f"{path}: invalid TOML: {exc}") from exc
    _check_keys(data, _TOP_KEYS, str(path))
    header = data.get("manifest")
    if not isinstance(header, dict):
        raise ManifestError(f"{path}: missing [manifest] table")
    _check_keys(header, _HEADER_KEYS, "manifest")
    scheduled = header.get("scheduled_wave")
    if not isinstance(scheduled, str) or not scheduled:
        raise ManifestError("manifest: 'scheduled_wave' is required")
    authoritative = _str_list(
        header.get("authoritative_waves"), "manifest.authoritative_waves"
    )
    raw_waves = data.get("waves")
    if not isinstance(raw_waves, dict) or not raw_waves:
        raise ManifestError(f"{path}: missing [waves] table")
    waves = {name: _str_list(roster, f"waves.{name}") for name, roster in raw_waves.items()}
    raw_tasks = data.get("tasks")
    if not isinstance(raw_tasks, dict) or not raw_tasks:
        raise ManifestError(f"{path}: missing [tasks.*] tables")
    tasks = {task_id: _parse_row(task_id, row) for task_id, row in raw_tasks.items()}
    return OwnershipManifest(
        path=path,
        authoritative_waves=authoritative,
        scheduled_wave=scheduled,
        waves=waves,
        tasks=tasks,
        plan=str(header.get("plan", "")),
    )