"""Load and resolve the config file; wire it into a runtime (2.3).
The pipeline, each step failing as one ``ConfigError`` with YAML paths::
raw text --(yaml.safe_load)--> dict
--(recursive "env:NAME" resolution from os.environ)--> dict
--(pydantic validation, schema.py)--> Config
--(build_runtime: the ONLY place config meets the backend
registry)--> Runtime
``env:NAME`` is how secrets and machine-specific values stay out of the file
(and out of git): any string value anywhere in the YAML may be ``env:NAME``,
replaced by ``os.environ["NAME"]`` before validation. An *unset* variable is a
hard error naming the variable and its YAML path -- it never resolves to an
empty string, because an empty string silently becoming a hostname or a
password is exactly the failure mode this exists to prevent.
``build_runtime`` turns the validated ``Config`` into live objects: it imports
the built-in backend modules (registration is explicit-import based), asks the
registry to construct each configured backend, and pre-computes the
``{stage_name: backend}`` mapping the orchestrator takes. Core never sees any
of this -- the dependency arrow points config -> backends, never backwards.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any
import yaml
from pydantic import ValidationError
from matmelt.backends.registry import create_backend
from matmelt.config.schema import Config
from matmelt.core.jobspec import BackendError
from matmelt.core.stages.base import StageError
from matmelt.core.stages.registry import FUNNEL_ORDER, OPTIONAL_STAGES, create_stage
if TYPE_CHECKING:
from matmelt.backends.base import ComputeBackend
from matmelt.core.stages.base import Stage
_ENV_PREFIX = "env:"
[docs]
class ConfigError(Exception):
"""Any problem loading, resolving, or validating the config file.
Always carries an actionable message with the YAML path of the offending
value (all validation problems are collected into one raise), so the user
fixes the file, not a stack trace.
"""
# ---------------------------------------------------------------------------
# Loading
# ---------------------------------------------------------------------------
[docs]
def load_config(path: str | os.PathLike[str]) -> Config:
"""Load the YAML file at ``path`` into a validated ``Config``.
Parameters
----------
path
The config file to read.
Returns
-------
Config
The fully-validated configuration.
Raises
------
ConfigError
For a missing, unreadable or empty file; invalid YAML; a non-mapping
top level; an unset ``env:`` reference; or any schema violation. Every
schema violation is reported **together**, one line each with its YAML
path, so a user fixes the whole file in one pass rather than one error
per run.
"""
raw = _read_yaml(Path(path))
resolved = _resolve_env(raw, "<root>")
try:
return Config.model_validate(resolved)
except ValidationError as exc:
raise ConfigError(_format_validation_error(exc, path)) from None
def _read_yaml(path: Path) -> dict[str, Any]:
"""Read and parse the file, refusing anything that is not a non-empty mapping.
Parameters
----------
path
The config file.
Returns
-------
dict
The raw parsed YAML, before ``env:`` resolution or validation.
Raises
------
ConfigError
For a missing or unreadable file, invalid YAML, an empty document, or
a top level that is not a mapping.
"""
if not path.is_file():
raise ConfigError(f"config file not found: {path}")
try:
text = path.read_text(encoding="utf-8")
except OSError as exc:
raise ConfigError(f"cannot read config file {path}: {exc}") from None
try:
data = yaml.safe_load(text)
except yaml.YAMLError as exc:
raise ConfigError(f"invalid YAML in {path}: {exc}") from None
if data is None:
raise ConfigError(f"config file {path} is empty")
if not isinstance(data, dict):
raise ConfigError(f"{path}: top level must be a mapping, got {type(data).__name__}")
return data
def _resolve_env(value: Any, path: str) -> Any:
"""Recursively replace every ``"env:NAME"`` string with the variable's value.
This is how a config file names a host or a credential without holding one:
no secret ever needs to be written into a file that might be committed.
Parameters
----------
value
The raw parsed YAML, or any node within it.
path
The dotted YAML path of ``value``, used in error messages.
Returns
-------
Any
The same structure with references resolved.
Raises
------
ConfigError
For a reference with no variable name, or a variable that is not set.
The message names the variable **and** its YAML path -- an unset
reference must never quietly become an empty string.
Notes
-----
Applied *before* validation, so a resolved value passes through exactly the
same typed checks as a literal one. pydantic coerces numeric strings, which
is what makes ``lease_timeout_s: env:LEASE`` work.
"""
if isinstance(value, dict):
return {key: _resolve_env(item, f"{path}.{key}") for key, item in value.items()}
if isinstance(value, list):
return [_resolve_env(item, f"{path}[{i}]") for i, item in enumerate(value)]
if isinstance(value, str) and value.startswith(_ENV_PREFIX):
name = value[len(_ENV_PREFIX) :]
if not name:
raise ConfigError(f"{path}: 'env:' reference has no variable name")
if name not in os.environ:
raise ConfigError(f"{path}: environment variable {name!r} (from {value!r}) is not set")
return os.environ[name]
return value
def _format_validation_error(exc: ValidationError, path: str | os.PathLike[str]) -> str:
"""Fold a pydantic ``ValidationError`` into one message, one line per problem.
Parameters
----------
exc
The validation error.
path
The config file's path, named in the summary line.
Returns
-------
str
A count, then one indented line per problem, each carrying the dotted
YAML path of the offending value.
"""
lines = []
for err in exc.errors():
loc = ".".join(str(part) for part in err["loc"]) or "<config>"
lines.append(f" - {loc}: {err['msg']}")
n = len(lines)
plural = "s" if n > 1 else ""
return f"{path}: {n} configuration error{plural}:\n" + "\n".join(lines)
# ---------------------------------------------------------------------------
# Wiring: config -> live objects
# ---------------------------------------------------------------------------
[docs]
@dataclass(frozen=True)
class Runtime:
"""The validated config plus the live backends built from it.
Attributes
----------
config : Config
The validated configuration.
backends : dict
One constructed ``ComputeBackend`` per entry under ``backends:``, keyed
by the user's configured name.
stage_backends : dict
The same instances re-keyed by *stage* name -- the exact
``{stage_name: backend}`` mapping the ``Orchestrator`` takes. Two stages
naming the same backend share one instance, and therefore share its
``max_parallel`` slots.
"""
config: Config
backends: dict[str, ComputeBackend]
stage_backends: dict[str, ComputeBackend]
[docs]
def build_runtime(config: Config) -> Runtime:
"""Construct every configured backend -- where config meets the registry.
Parameters
----------
config
The validated configuration.
Returns
-------
Runtime
The config, the constructed backends, and the stage-to-backend mapping.
Raises
------
ConfigError
For an unknown kind or an option a backend rejects, carrying the
entry's YAML path. At this point such a failure is a *configuration*
problem, not an infrastructure one, which is why it is re-raised rather
than left as a ``BackendError``.
Notes
-----
The built-in backend modules are imported here so their kinds register.
Registration is an explicit call at the bottom of each module, never
import-time magic, so there is no hidden ordering dependency.
"""
_import_builtin_backends()
backends: dict[str, ComputeBackend] = {}
for name, backend_cfg in config.backends.items():
try:
backends[name] = create_backend(name, backend_cfg.kind, backend_cfg.options)
except BackendError as exc:
raise ConfigError(f"backends.{name}: cannot construct backend: {exc}") from None
stage_backends = {
stage_name: backends[stage_cfg.backend] for stage_name, stage_cfg in config.stages.items()
}
return Runtime(config=config, backends=backends, stage_backends=stage_backends)
def _import_builtin_backends() -> None:
"""Import the backend modules this tool ships, so their kinds register.
Registration happens at the bottom of each concrete backend module, so
importing here is what makes those kinds available to a config file. A
third-party backend registers the same way, from wherever it is imported.
"""
import matmelt.backends.local # registers "local"
import matmelt.backends.ssh_exec # noqa: F401 (registers "ssh_exec")
[docs]
def build_stages(config: Config) -> list[Stage]:
"""Construct the configured stages in funnel order -- config meets the registry.
Parameters
----------
config
The validated configuration.
Returns
-------
list of Stage
Exactly what the ``Orchestrator`` takes: each stage consumes the
previous one's output state.
Raises
------
ConfigError
For an unknown stage name; for a configured set that does not follow
the funnel; or for an option a stage does not accept or rejects,
carrying the entry's YAML path.
Notes
-----
The configured names must form an ordered subsequence of the funnel
``compose -> structure -> screen -> [calculate] -> extract``, omitting only
optional stages. Every required stage up to the furthest configured one
must be present: a funnel starting mid-chain would never consume ``NEW``,
and a hole at a required stage is refused rather than silently bridged.
The YAML order of the ``stages:`` entries is cosmetic -- it is a mapping.
The funnel defines the order.
"""
_import_builtin_stages()
for name in config.stages:
if name not in FUNNEL_ORDER:
raise ConfigError(
f"stages.{name}: unknown stage; available stages, in funnel order: "
f"{', '.join(FUNNEL_ORDER)}"
)
ordered = [name for name in FUNNEL_ORDER if name in config.stages]
last_index = FUNNEL_ORDER.index(ordered[-1])
missing = [
name
for name in FUNNEL_ORDER[:last_index]
if name not in config.stages and name not in OPTIONAL_STAGES
]
if missing:
funnel = " -> ".join(
f"[{name}]" if name in OPTIONAL_STAGES else name for name in FUNNEL_ORDER
)
raise ConfigError(
f"stages: the configured stages must follow the funnel ({funnel}), "
f"omitting only optional stages; got [{', '.join(sorted(config.stages))}] "
f"-- missing [{', '.join(missing)}]"
)
stages: list[Stage] = []
for name in ordered:
try:
stages.append(create_stage(name, config.stages[name].options))
except StageError as exc:
raise ConfigError(f"stages.{name}: cannot construct stage: {exc}") from None
return stages
def _import_builtin_stages() -> None:
"""Import the stage modules this tool ships, so their names register.
Registration happens at the bottom of each concrete stage module, so
importing here is what makes those stages available to a config file.
Importing a stage module costs nothing scientific: every science dependency
is imported lazily inside its runner, never at module scope.
"""
import matmelt.core.stages.calculate # registers "calculate"
import matmelt.core.stages.compose # registers "compose"
import matmelt.core.stages.extract # registers "extract"
import matmelt.core.stages.screen # registers "screen"
import matmelt.core.stages.structure # noqa: F401 (registers "structure")