|
| 1 | +# SPDX-FileCopyrightText: 2025 Alexandre Gomes Gaigalas <alganet@gmail.com> |
| 2 | +# SPDX-License-Identifier: ISC |
| 3 | + |
| 4 | +"""Core wiring functionality.""" |
| 5 | + |
| 6 | +import importlib |
| 7 | +from typing import Dict, List, Tuple, TypeAlias, TypeVar, cast |
| 8 | + |
| 9 | +T = TypeVar("T") |
| 10 | + |
| 11 | +InstanceData: TypeAlias = Dict[str, object] |
| 12 | +Blueprint: TypeAlias = Tuple[str, str, str, InstanceData] |
| 13 | + |
| 14 | +Spec: TypeAlias = Dict[str, InstanceData] |
| 15 | + |
| 16 | + |
| 17 | +def _parse_spec[T](spec: Spec) -> List[Blueprint]: |
| 18 | + """Parses a Spec into Blueprints for instantiation.""" |
| 19 | + items = [] |
| 20 | + for key, value in spec.items(): |
| 21 | + type_str, name = key.rsplit(" ", 1) |
| 22 | + parts = type_str.split(".") |
| 23 | + module_name = ".".join(parts[:-1]) |
| 24 | + class_name = parts[-1] |
| 25 | + items.append((module_name, class_name, name, value)) |
| 26 | + return items |
| 27 | + |
| 28 | + |
| 29 | +class Wired[T]: |
| 30 | + """Lazy-loaded container for wired objects.""" |
| 31 | + |
| 32 | + def __init__(self, parsed: List[Blueprint]) -> None: |
| 33 | + self._parsed = { |
| 34 | + name: (module_name, class_name, value) |
| 35 | + for module_name, class_name, name, value in parsed |
| 36 | + } |
| 37 | + self._instances: Dict[str, object] = {} |
| 38 | + |
| 39 | + def __getattr__(self, name: str) -> object: |
| 40 | + """Get the instantiated object for name via attribute access.""" |
| 41 | + if name not in self._parsed: |
| 42 | + raise AttributeError(f"no attribute '{name}'") |
| 43 | + if name not in self._instances: |
| 44 | + module_name, class_name, value = self._parsed[name] |
| 45 | + module = importlib.import_module(module_name) |
| 46 | + cls = cast(type[T], getattr(module, class_name)) |
| 47 | + self._instances[name] = cls(**value) |
| 48 | + return self._instances[name] |
| 49 | + |
| 50 | + |
| 51 | +def wire[T](spec: Spec) -> Wired[T]: |
| 52 | + """Creates a lazy-loaded Wired object from a Spec.""" |
| 53 | + return Wired(_parse_spec(spec)) |
| 54 | + |
| 55 | + |
| 56 | +def compile[T](spec: Spec) -> str: |
| 57 | + """Compiles a Spec into a string containing Python code.""" |
| 58 | + parsed = _parse_spec(spec) |
| 59 | + modules = set() |
| 60 | + properties = [] |
| 61 | + for module_name, class_name, name, value in parsed: |
| 62 | + modules.add(module_name) |
| 63 | + type_str = f"{module_name}.{class_name}" |
| 64 | + if isinstance(value, dict): |
| 65 | + kwargs = ", ".join(f"{k}={repr(v)}" for k, v in value.items()) |
| 66 | + prop = ( |
| 67 | + " @property\n" |
| 68 | + f" def {name}(self):\n" |
| 69 | + f" return {type_str}({kwargs})" |
| 70 | + ) |
| 71 | + properties.append(prop) |
| 72 | + imports = [f"import {module}" for module in sorted(modules)] |
| 73 | + class_def = ["class Compiled:"] + properties + ["compiled = Compiled()"] |
| 74 | + return "\n".join(imports + class_def) |
0 commit comments