API Reference

The engine and its context objects. Rendered from source via mkdocstrings.

Engine

netboot.Pixie

Source code in src/netboot/__init__.py

class Pixie:
    targets: "dict[str,PixieTarget]"
    dhcpzones: "dict[str,DhcpZone]"
    images: "dict[str, PixieImage]"
    repos: "dict[str,Repository]"
    globals: dict[str, object]
    _ctxcls: PixieContext = PixieContext
    VERSION = __version__
    _config: dict
    _hooks: list[_PixieHook] = []

    def hook(
        self: "Pixie|_ty.Sequence[_PixieHook]",
        event: PixieEvent,
        __value: T = None,
        /,
        **kwargs,
    ):
        LOGGER.debug(f"Running Hooks for: {event}")
        if isinstance(self, Pixie):
            netboot = self
            hooks = self._hooks
        else:
            netboot = None
            hooks = self
        for hook in hooks:
            __value = hook(event, netboot, __value, kwargs)
        return __value

    def __new__(
        cls,
        /,
        hooks: _ty.Sequence[_PixieHook] = [],
        **config,
    ):

        hooks = [(hook if callable(hook) else import_(hook)) for hook in hooks]
        cls = Pixie.hook(hooks, PixieEvent.NewPixieObject, cls, config=config)
        inst = object.__new__(cls)
        inst._hooks = hooks
        return inst

    def __init__(
        netboot,
        /,
        **config,
    ):
        netboot._config = netboot.hook(PixieEvent.StartPixieInit, config)
        netboot.globals = deepcopy(config.get("globals") or {})
        defaults = config.get("defaults") or {}

        for prop, hint in get_type_hints(netboot.__class__).items():
            if prop.startswith("_"):
                continue
            origin = _ty.get_origin(hint) or hint
            value = config.get(prop)
            prop_defaults = defaults.get(prop)
            if issubclass(origin, dict):
                value: dict[str] = value or {}
                _keycls, _valcls = _ty.get_args(hint)
                _value = {}
                if _valcls is object:
                    _valctr = lambda _id, val: val
                else:

                    def _valctr(_id, val):
                        val = val if isinstance(val, Mapping) else val.__dict__
                        if prop_defaults:
                            _val = deepcopy(prop_defaults)
                            _val.update(val)
                            val = _val
                        try:
                            return _valcls(_id=_id, **val)
                        except TypeError:
                            # _valcls doesn't accept an _id kwarg; build without.
                            return _valcls(**val)

                for uid, val in value.items():
                    if not str(uid).startswith("_"):
                        _value[_keycls(uid)] = _valctr(uid, val)
            else:
                _value = hint(value) if value is not None else None
            prop, value = netboot.hook(
                PixieEvent.SetPixieProperty, (prop, _value), origin=origin, rawvalue=value
            )
            setattr(netboot, prop, value)

        netboot.hook(PixieEvent.PixieInitiated)

    def lookup_target(self, target: str) -> PixieTarget:
        target: Union[str, PixieTarget] = self.hook(PixieEvent.LookupTarget, target)
        if isinstance(target, str):
            lower: str = target.lower()
            _target = self.targets.get(target, None)
            if _target is None:
                for _target in self.targets.values():
                    if (
                        _target.hostname.lower().startswith(lower)
                        or _target.mac.as_str() == lower
                        or str(_target.ip) == lower
                    ):
                        target = _target
                        break

            else:
                target = _target

        target = self.hook(PixieEvent.FoundTarget, target)
        if not isinstance(target, PixieTarget):
            target = None
        return target

    def lookup_image(self, name: str, target: PixieTarget = None) -> PixieImage:
        imgs: list[tuple[int, PixieImage]] = [(-1, {})]
        for img_name, image in self.images.items():
            check = image.match(img_name, name)
            if check != False:
                imgs.append((check, image))
        imgs.sort(key=lambda x: x[0])
        return self.hook(PixieEvent.FoundTargetImage, imgs.pop()[1], target=target)

    def lookup_dhcpzone(self, name: str, target: PixieTarget = None) -> DhcpZone:
        if not name and target:
            if target.dhcpzone:
                name = target.dhcpzone
            else:
                for zone_id, zone in self.dhcpzones.items():
                    if target.ip in zone.network:
                        name = zone_id
                        target.dhcpzone = zone_id
                        break
        zone = self.dhcpzones.get(name, None)
        return self.hook(PixieEvent.FoundTargetDhcpzone, zone, target=target)

    def make_context(
        self, target: "PixieTarget", globals: list[dict] = None
    ) -> PixieContext:
        image = self.lookup_image(target.image, target)
        if not image:
            raise Exception(f"Image not found for target: {target.image}")
        LOGGER.info(f"Found target image: {target.image}")

        zone = self.lookup_dhcpzone(target.dhcpzone, target)
        if not zone:
            raise Exception(
                f"Not supported client due to missing subnet: {target.dhcpzone}"
            )
        LOGGER.info(f"Found target dhcpzone: {target.dhcpzone}")

        ctx = {
            "image": image,
            "dhcpzone": zone,
            "target": target,
            "repos": self.repos,
            "resources": {},
        }
        # Collect object-level globals to layer into the context WITHOUT mutating
        # the shared image/dhcpzone/target objects (they are reused across
        # targets, so deleting their `globals` would drop them for later calls).
        _globals = [self.globals, *(globals or [])]
        for k in ["image", "dhcpzone", "target"]:
            g = getattr(ctx[k], "globals", None)
            if g:
                _globals.append(g)

        ctx = mergeObjects(self._ctxcls, ctx, *_globals)

        ctx: PixieContext
        ctx._renderer = Renderer(loader=Loader(self._config.get("templates", [])))
        ctx.version = f"netboot-v{self.VERSION}"
        return self.hook(PixieEvent.PixieContextForTarget, ctx, target=target)

    def initialize(self, target: "PixieTarget"):
        target = self.hook(PixieEvent.StartPixieInitialize, target)
        ctx = self.make_context(target)
        ctx = ctx.pxe_init(self)
        return self.hook(PixieEvent.EndPixieInitialize, ctx)

    def complete(self, target: "PixieTarget"):
        target = self.hook(PixieEvent.StartPixieComplete, target)
        ctx = self.make_context(target)
        ctx = ctx.pxe_complete(self)
        return self.hook(PixieEvent.EndPixieComplete, ctx)

Context

netboot.PixieContext

Bases: Namespace

Source code in src/netboot/__init__.py

class PixieContext(Namespace):
    target: PixieTarget
    image: PixieImage
    dhcpzone: DhcpZone
    repos: dict[str, Repository]
    generated: datetime.datetime
    resources: dict[str, Resource]
    version: str
    templates: list[Union[str, Path]]
    _renderer: Renderer

    def __init__(self, **kwargs) -> None:
        self.dhcp_server = None
        self.generated = datetime.datetime.now()
        super().__init__(**kwargs)

    def init(self, netboot: "Pixie"): ...

    def resource(self, name: Union[str, Resource], service: str = None):
        if isinstance(name, Resource):
            repo = self.repos.get(name.src, None)
            path = name.path
        else:
            path = None
            repo = self.resource_repo(name)
        if not repo:
            return
        if not path:
            path = self.resources[name].path
        return repo.get(path, service=service)

    def resource_repo(self, name: str):
        resource = self.resources.get(name)
        if not resource:
            return
        return self.repos.get(resource.src)

    def pxe_init(self, config: "Pixie"):
        for dhcpserver in self.dhcpzone.dhcpservers:
            dhcpserver.add_target(self)
        return self

    def pxe_complete(self, config: "Pixie"):
        for dhcpserver in self.dhcpzone.dhcpservers:
            dhcpserver.remove_target(self)
        return self

    def _template_names(self, suffix: Union[list[str], str], **options) -> list[str]:
        # ``options`` (a ``k=v:name`` template spec) is accepted for forward
        # compatibility; name selection does not use it today.
        suffixes = suffix if isinstance(suffix, list) else [suffix]
        ip = self.target.ip
        # Skip an unset/unspecified IP so it never yields a spurious name.
        ip_name = str(ip) if ip and str(ip) not in ("0.0.0.0", "::") else ""
        names = []
        for version in [
            self.target.mac.as_str("-"),
            self.target.hostname,
            ip_name,
        ]:
            if version:
                for suffix in suffixes:
                    names.append(f"{version}.{suffix}")
        for suffix in suffixes:
            names.append(suffix)

        return names

    @property
    def searchpaths(self) -> list[Path]:
        return [*self.target.template_path, *self.image.template_path]

    def render(self, filename: str, strict=True):
        # Each PixieContext owns its own Renderer (built in make_context), so
        # setting globals["ctx"] here is per-context; do not share one Renderer
        # across contexts or nested renders would clobber this.
        self._renderer.globals["ctx"] = self
        try:
            template = self._renderer.get_template(filename)
            return template.render()
        except Exception as e:
            if strict:
                raise e
            else:
                LOGGER.debug(
                    f"While rendering [{filename}] encountered an exeption:\t{repr(e)}"
                )
                return None

Targets and images

netboot.PixieTarget

Bases: Namespace, OpaqueMerge

Source code in src/netboot/__init__.py

class PixieTarget(Namespace, OpaqueMerge):
    _id: str
    hostname: str
    ip: IPAddress
    mac: MACAddress
    image: str
    dhcpzone: str
    globals: dict
    template_path: list[Union[str, Path]]

    #: MAC value treated as "unset" (a target keyed by hostname/IP has no MAC).
    _NULL_MAC = "00:00:00:00:00:00"

    def __init__(self, **kwargs) -> None:
        for prop, key in {
            "dhcpzone": "",
            "mac": "",
            "ip": "",
            "image": "",
            "hostname": "",
            "template_path": [],
        }.items():
            kwargs.setdefault(prop, key)
        super().__init__(**kwargs)
        # If no MAC was given but the id itself is a MAC, adopt it; otherwise
        # fall back to the null MAC so downstream `.mac` is always a MACAddress.
        if not self.mac or str(self.mac) == self._NULL_MAC:
            if MACAddress._VALID_MAC.match(str(self._id)):
                self.mac = self._id
            else:
                self.mac = self._NULL_MAC
        if not isinstance(self.mac, MACAddress):
            self.mac = MACAddress(self.mac)
        resolve = not self.ip or not self.hostname
        while resolve:
            resolve = False
            not_mac = not MACAddress._VALID_MAC.match(self._id)
            id_is_ip = netutils.is_valid_ip(self._id)
            if not self.ip and self.hostname:
                _resolved = netutils.nslookup(self.hostname)
                if _resolved:
                    self.ip = _resolved[0]
                resolve = True
            if not self.ip and id_is_ip:
                self.ip = self._id
                resolve = True
            if not self.hostname and not_mac and not id_is_ip:
                self.hostname = self._id
                resolve = True
        if self.ip:
            self.ip = netutils.parse_ip(self.ip)
        self.hostname = self.hostname.lower()

netboot.PixieImage

Bases: Resource

Source code in src/netboot/__init__.py

class PixieImage(Resource):
    template_path: list["Path"]
    globals: dict

    def match(self, name: str, check: str):
        return name == check

Events

netboot.PixieEvent

Bases: StrEnum

Source code in src/netboot/__init__.py

class PixieEvent(StrEnum):
    NewPixieObject = "PixieEvent.NewPixieObject"
    StartPixieInit = "PixieEvent.StartPixieInit"
    SetPixieProperty = "PixieEvent.SetPixieProperty"
    PixieInitiated = "PixieEvent.PixieInitiated"
    LookupTarget = "PixieEvent.LookupTarget"
    FoundTarget = "PixieEvent.FoundTarget"
    FoundTargetImage = "PixieEvent.FoundTargetImage"
    FoundTargetDhcpzone = "PixieEvent.FoundTargetDhcpzone"
    PixieContextForTarget = "PixieEvent.PixieContextForTarget"
    StartPixieInitialize = "PixieEvent.StartPixieInitialize"
    EndPixieInitialize = "PixieEvent.EndPixieInitialize"
    StartPixieComplete = "PixieEvent.StartPixieComplete"
    EndPixieComplete = "PixieEvent.EndPixieComplete"

DHCP

netboot.dhcp.DhcpZone

Bases: Namespace, OpaqueMerge

Source code in src/netboot/dhcp.py

class DhcpZone(Namespace, OpaqueMerge):
    network: IPNetwork
    gateway: "_ty.Optional[IPAddress]"
    domain: "_ty.Optional[str]"
    search: "list[str]"
    nameservers: "list[IPAddress]"
    globals: dict
    dhcpservers: "list[DhcpServer]"

    @property
    def nameserver(self):
        return self.nameservers[0] if self.nameservers else ""

    def get_local_server(self, servers: "list[IPAddress]", default: IPAddress):
        for server in servers:
            if server in self.network:
                return server
        return default

    def __init__(self, **kwargs) -> None:
        _gateway = kwargs.get("gateway")
        _network = kwargs.get("network")
        _netmask = kwargs.get("netmask")
        if _gateway and not _network:
            gw: netutils.IPv4Interface = IPInterface(_gateway)
            if not _netmask and isinstance(_gateway, str) and "/" in _gateway:
                _netmask = gw.netmask.exploded
                kwargs["gateway"] = gw.ip.exploded
            if _netmask:
                _network = gw.network.network_address.exploded

        if isinstance(_network, str) and "/" in _network:
            net = IPNetwork(_network)
            _network = net.network_address.exploded
            _netmask = net.netmask.exploded
        if _network:
            if _netmask and isinstance(_network, str):
                _network = f"{_network}/{_netmask}"
            kwargs["network"] = _network

        kwargs.pop("netmask", None)
        kwargs.setdefault("dhcpservers", [])
        super().__init__(**kwargs)
        self.dhcpservers = [
            server if isinstance(server, DhcpServer) else DhcpServer(server)
            for server in (self.dhcpservers or [])
        ]
        self.network = netutils.parse_network(self.network)
        for prop, ctr in [("nameservers", netutils.parse_ip), ("search", str)]:
            value = getattr(self, prop, None)
            if not value:
                setattr(self, prop, [])
            elif not isinstance(value, list):
                setattr(self, prop, [value])
            vals = getattr(self, prop)
            for i, val in enumerate(vals):
                vals[i] = ctr(val)

        for prop in ["gateway", "domain"]:
            if not getattr(self, prop, None):
                setattr(self, prop, None)

        if self.gateway:
            self.gateway = netutils.parse_ip(self.gateway)

netboot.dhcp.DhcpServer

Base for DHCP backends. DhcpServer(uri) dispatches on the URI scheme.

A concrete backend is a subclass whose lowercased class name matches the URI scheme (e.g. class dnsmasq(DhcpServer) handles dnsmasq://...); such backends are typically provided by a plugin module imported via --load-module. Subclassing at any depth is honoured.

Source code in src/netboot/dhcp.py

class DhcpServer:
    """Base for DHCP backends. ``DhcpServer(uri)`` dispatches on the URI scheme.

    A concrete backend is a subclass whose lowercased class name matches the URI
    scheme (e.g. ``class dnsmasq(DhcpServer)`` handles ``dnsmasq://...``); such
    backends are typically provided by a plugin module imported via
    ``--load-module``. Subclassing at any depth is honoured.
    """

    DEFAULT_SCHEME = None

    def __new__(cls, uri: str):
        if cls is DhcpServer:
            parsed = urlparse(uri)
            scheme = parsed.scheme or cls.DEFAULT_SCHEME
            for sub in _iter_subclasses(cls):
                if sub.__name__.lower() == scheme:
                    return object.__new__(sub)
            raise ValueError(
                f"No DhcpServer backend registered for scheme {scheme!r} "
                f"(uri={uri!r}); import a plugin module providing it via --load-module"
            )
        return object.__new__(cls)

    def __init__(self, uri: str):
        self.uri = uri

    def remove_target(self, netboot: "PixieContext"):
        pass

    def add_target(self, netboot: "PixieContext"):
        pass