93 lines
3.0 KiB
Python
93 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
from types import ModuleType
|
|
from typing import Any
|
|
from dataclasses import asdict
|
|
|
|
from retrobuilder.model import BaseSpec, FeatureSpec, ProfileSpec
|
|
from retrobuilder.paths import base_dir, feature_dir, profile_file
|
|
|
|
|
|
def _load_module(path: Path, name: str) -> ModuleType:
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError(f"Cannot load module from {path}")
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def load_profile(root: Path, name: str) -> ProfileSpec:
|
|
module = _load_module(profile_file(root, name), f"profile_{name}")
|
|
profile = getattr(module, "PROFILE", None)
|
|
if not isinstance(profile, ProfileSpec):
|
|
raise TypeError(f"profiles/{name}.py must define PROFILE as ProfileSpec")
|
|
return profile
|
|
|
|
|
|
def _load_module_py(root: Path, kind: str, name: str) -> ModuleType:
|
|
path = (base_dir if kind == "base" else feature_dir)(root, name) / "entry.py"
|
|
return _load_module(path, f"{kind}_{name}")
|
|
|
|
|
|
def load_module_spec(root: Path, kind: str, name: str):
|
|
module = _load_module_py(root, kind, name)
|
|
spec = getattr(module, "SPEC", None)
|
|
expected = BaseSpec if kind == "base" else FeatureSpec
|
|
if not isinstance(spec, expected):
|
|
raise TypeError(f"{kind}s/{name}/entry.py must define SPEC as {expected.__name__}")
|
|
return spec
|
|
|
|
|
|
def load_module_entry(root: Path, kind: str, name: str):
|
|
module = _load_module_py(root, kind, name)
|
|
entry = getattr(module, "Entry", None)
|
|
if entry is None:
|
|
raise TypeError(f"{kind}s/{name}/entry.py must define Entry")
|
|
return entry
|
|
|
|
|
|
def load_base_chain(root: Path, name: str):
|
|
return load_module_chain(root, "base", name)
|
|
|
|
|
|
def load_module_chain(root: Path, kind: str, name: str):
|
|
chain: list[tuple[str, Any]] = []
|
|
seen: set[str] = set()
|
|
current = name
|
|
while current:
|
|
if current in seen:
|
|
raise RuntimeError(f"{kind} inheritance loop detected at {current}")
|
|
seen.add(current)
|
|
spec = load_module_spec(root, kind, current)
|
|
chain.append((current, spec))
|
|
current = spec.parent or ""
|
|
chain.reverse()
|
|
return chain
|
|
|
|
|
|
def resolve_module_config(root: Path, kind: str, name: str) -> dict[str, Any]:
|
|
merged: dict[str, Any] = {}
|
|
for _name, spec in load_module_chain(root, kind, name):
|
|
data = asdict(spec)
|
|
|
|
data.pop("description", None)
|
|
data.pop("parent", None)
|
|
data.pop("docker_overrides", None)
|
|
extra = data.pop("config", {}) or {}
|
|
|
|
merged.update(data)
|
|
merged.update(extra)
|
|
|
|
return merged
|
|
|
|
|
|
def resolve_module_docker_override(root: Path, kind: str, name: str, phase: str):
|
|
resolved = None
|
|
for _name, spec in load_module_chain(root, kind, name):
|
|
if phase in spec.docker_overrides:
|
|
resolved = spec.docker_overrides[phase]
|
|
return resolved
|