Source code for codegen.config

import json
from typing import TYPE_CHECKING, Any

from pydantic import BaseModel, ConfigDict, ValidationError

from codegen import jsonnet, jsonpath
from codegen.errors import ConfigError
from codegen.plugins import load_plugins

if TYPE_CHECKING:
    from collections.abc import Mapping
    from pathlib import Path

    from codegen.plugins import JsonnetTarget
    from codegen.target import Target




class ExtensibleConfig(BaseModel):
    model_config = ConfigDict(extra="allow")

    @property
    def options(self) -> dict[str, Any]:
        return self.model_extra or {}





class CodegenConfig(BaseModel):
    package_prefix: str = ""



def load_config(
    path: Path,
    target: Target,
    stdlibs: Mapping[str, Path] | None = None,
) -> CodegenConfig:
    compiled_json = json.loads(_read_source(path, stdlibs or {}))

    for plugin in load_plugins():
        for jsonnet_target in plugin.jsonnet_targets:
            if target.name in jsonnet_target.targets:
                plugin_fragment = json.loads(
                    jsonnet.evaluate(
                        path=plugin.jsonnet_dir() / jsonnet_target.path,
                        stdlibs=stdlibs or {},
                    )
                )
                _insert_jsonnet_at_jsonpath(
                    compiled_json, jsonnet_target, plugin_fragment
                )

    try:
        return target.schema.model_validate(compiled_json)

    except ValidationError as exc:
        msg = f"Invalid config in {path}: {exc}"
        raise ConfigError(msg) from exc


def _insert_jsonnet_at_jsonpath(
    data: dict,
    target: JsonnetTarget,
    fragment: list[Any],
) -> None:
    matches = jsonpath.search(target.jsonpath, data)

    if not matches:
        msg = f"{target.path}: jsonpath {target.jsonpath!r} matched nothing"
        raise ConfigError(msg)

    if not target.allows_many and len(matches) > 1:
        msg = (
            f"{target.path}: jsonpath {target.jsonpath!r} has too many matches"
        )
        raise ConfigError(msg)

    for match in matches:
        if not isinstance(match, list):
            msg = (
                f"{target.path}: {target.jsonpath!r} selected a "
                f"{type(match).__name__}, expected a list"
            )
            raise ConfigError(msg)

        match.extend(fragment)


def _read_source(path: Path, stdlibs: Mapping[str, Path]) -> str:
    suffix = path.suffix.lower()

    try:
        if suffix == ".jsonnet":
            return jsonnet.evaluate(path, stdlibs)

        if suffix == ".json":
            return path.read_text()

    except RuntimeError as exc:
        msg = f"Jsonnet evaluation failed for {path}: {exc}"
        raise ConfigError(msg) from exc

    except OSError as exc:
        msg = f"Could not read {path}: {exc}"
        raise ConfigError(msg) from exc

    msg = f"Unsupported config format: {suffix!r} (use .json or .jsonnet)"
    raise ConfigError(msg)