API Reference

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

Engine

netboot.Pixie

The engine: config in, provisioning decisions out.

Built from a merged config mapping, it resolves targets, images and DHCP zones, builds render contexts, and arms or disarms DHCP through the initialize/complete lifecycle. Every step fires a PixieEvent, so behaviour can be changed without subclassing.

Source code in src/netboot/engine.py

class Pixie:
    """The engine: config in, provisioning decisions out.

    Built from a merged config mapping, it resolves targets, images and DHCP
    zones, builds render contexts, and arms or disarms DHCP through the
    `initialize`/`complete` lifecycle. Every step fires a `PixieEvent`, so
    behaviour can be changed without subclassing.
    """

    targets: "dict[str,PixieTarget]"
    dhcpzones: "dict[str,DHCPZone]"
    images: "dict[str, PixieImage]"
    repos: "dict[str,Repository]"
    globals: dict[str, object]
    #: What a template variable with no value does: "strict" raises naming it,
    #: "lenient" renders an empty string, "debug" leaves the placeholder in the
    #: output. Set from the config key of the same name; it means the same in
    #: both template engines.
    templates_undefined: str = "strict"
    _ctxcls: "type[PixieContext]" = PixieContext
    VERSION = __version__
    _config: dict
    _hooks: list[_PixieHook] = []

    def hook(
        self: "Pixie|_ty.Sequence[_PixieHook]",
        event: PixieEvent,
        __value: T = None,
        /,
        **kwargs: object,
    ) -> T:
        """Run the hook chain for `event`, threading `value` through it.

        Each hook is called as `f(event, netboot, value, kwargs)` and **must
        return** the value to pass on. `value` is positional-only.
        """
        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.Optional[_ty.Sequence[_PixieHook]]" = (),
        **config: object,
    ) -> "Pixie":

        hooks = [(hook if callable(hook) else import_(hook)) for hook in hooks or ()]
        cls = Pixie.hook(hooks, PixieEvent.NewPixieObject, cls, config=config)
        if not isinstance(cls, type) or not issubclass(cls, Pixie):
            # The event threads the class through the chain, so a hook that
            # forgets to return it replaces the engine class with None.
            raise PixieConfigError(
                "a NewPixieObject hook must return a Pixie subclass; got " f"{cls!r}"
            )
        inst = object.__new__(cls)
        inst._hooks = hooks
        return inst

    def __init__(
        netboot,
        /,
        **config: object,
    ) -> None:
        # Use what the hook returned, everywhere. Storing it while reading
        # the original meant a hook that returned a new mapping (the natural
        # non-mutating style) had its targets ignored but its templates
        # honoured -- the config ended up half applied.
        config = netboot._config = netboot.hook(PixieEvent.StartPixieInit, config)
        if config is None:
            raise PixieConfigError(
                "a StartPixieInit hook returned None; it must return the config "
                "mapping to pass on"
            )
        _warn_unknown_keys(
            "", config, _config_keys(netboot.__class__) | _TOP_LEVEL_KEYS
        )
        netboot.globals = deepcopy(config.get("globals") or {})
        # Read before the zones, whose servers' URIs may use the names declared.
        netboot.option_codes = codes_for(config.get("dhcp"))
        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 prop == "globals":
                # Keep the deep copy made above -- rebuilding it here would
                # undo the isolation, and would drop `_`-prefixed entries,
                # which in globals are ordinary variable names rather than the
                # collection ids that rule is meant for.
                _value = netboot.globals
            # `isinstance(origin, type)` first: a union annotation's origin is
            # not a class at all on 3.9, and `issubclass` raises on it.
            elif isinstance(origin, type) and issubclass(origin, dict):
                value: dict[str] = {} if value is None else value
                if not isinstance(value, Mapping):
                    raise PixieConfigError(
                        f"{prop}: expected a mapping of id to entry, got "
                        f"{type(value).__name__}"
                    )
                _keycls, _valcls = _ty.get_args(hint)
                _known = None if _valcls is object else _config_keys(_valcls)
                if _known is not None and isinstance(prop_defaults, Mapping):
                    _warn_unknown_keys(f"defaults.{prop}.", prop_defaults, _known)
                _value = {}
                if _valcls is object:
                    _valctr = lambda _id, val: val
                else:

                    def _valctr(_id, val):
                        # `targets: {host1:}` is valid YAML and means None; it
                        # is also the natural way to write "this entry, all
                        # defaults", so treat it as an empty mapping instead of
                        # failing on `None.__dict__`.
                        if val is None:
                            val = {}
                        elif not isinstance(val, Mapping):
                            fields = getattr(val, "__dict__", None)
                            if fields is None:
                                raise PixieConfigError(
                                    f"{prop}.{_id}: expected a mapping, got "
                                    f"{type(val).__name__} ({val!r})"
                                )
                            val = fields
                        if _known is not None:
                            _warn_unknown_keys(f"{prop}.{_id}.", val, _known)
                        if prop_defaults:
                            _val = deepcopy(prop_defaults)
                            _val.update(val)
                            val = _val
                        if isinstance(_valcls, type) and issubclass(_valcls, DHCPZone):
                            val = dict(val, codes=netboot.option_codes)
                        try:
                            return _valcls(_id=_id, **val)
                        except TypeError as exc:
                            # Only "this class takes no _id" justifies a retry;
                            # any other TypeError is the value class's own and
                            # must not be hidden behind a second construction.
                            if "_id" not in str(exc):
                                raise
                            return _valcls(**val)

                for uid, val in value.items():
                    if not isinstance(uid, str):
                        raise PixieConfigError(
                            f"{prop}: the id {uid!r} is not text "
                            f"({type(uid).__name__}); quote it (YAML reads an unquoted "
                            "52:54:00:12:34:56 as a number, and yes as a boolean)"
                        )
                    if uid.startswith("_"):
                        continue
                    try:
                        _value[_keycls(uid)] = _valctr(uid, val)
                    except PixieError:
                        raise
                    except ValueError as exc:  # a MAC, address or network
                        raise PixieConfigError(f"{prop}.{uid}: {exc}") from exc
            elif value is None:
                # Leave a class-level default in place instead of overwriting
                # it with None just because the config does not mention it.
                _value = getattr(netboot, prop, None)
            else:
                # Coerce through the annotation when it is a real constructor.
                # `Optional[str]` is not one: under PEP 604 its origin is
                # `types.UnionType`, which passes `isinstance(..., type)` and
                # then refuses to be called.
                try:
                    _value = origin(value) if isinstance(origin, type) else value
                except TypeError:
                    _value = value
            prop, value = netboot.hook(
                PixieEvent.SetPixieProperty,
                (prop, _value),
                origin=origin,
                rawvalue=value,
            )
            setattr(netboot, prop, value)

        netboot.hook(PixieEvent.PixieInitiated)

    def _match_target(self, query: str) -> "PixieTarget|None":
        """Find the target a user means by id, hostname, MAC or IP.

        An exact match anywhere in the table beats a prefix match, and an
        ambiguous query raises instead of picking whichever entry comes first:
        guessing here arms PXE on a machine nobody asked for. MAC input is
        parsed, so colon, hyphen and Cisco-dot spellings all match.
        """
        if not query:
            return None
        exact = self.targets.get(query, None)
        if exact is not None:
            return exact

        lower = query.lower()
        mac = netimps.try_parse(query, MACAddress)
        if mac is not None and str(mac) == PixieTarget._NULL_MAC:
            mac = None  # the "unset" MAC must not match every MAC-less target
        ip = netimps.try_parse(query, IPAddress)

        def exact():
            return [
                candidate
                for candidate in self.targets.values()
                if candidate.hostname.lower() == lower
                or (mac is not None and candidate.mac == mac)
                or (ip is not None and candidate.ip == ip)
            ]

        matches = exact()
        if not matches and ip is not None:
            # The address may be a hostname-only target's: look those up now.
            for candidate in self.targets.values():
                candidate.resolve_ip()
            matches = exact()
        if not matches:
            matches = [
                candidate
                for candidate in self.targets.values()
                if candidate.hostname and candidate.hostname.lower().startswith(lower)
            ]
        if len(matches) > 1:
            raise PixieLookupError(
                f"ambiguous target {query!r}: matches "
                f"{sorted(str(candidate._id) for candidate in matches)}"
            )
        return matches[0] if matches else None

    def lookup_target(self, target: str) -> "PixieTarget|None":
        """Find the target a user named, or `None`.

        Exact id first, then an exact hostname/MAC/IP match anywhere in the table,
        then a unique hostname prefix. Raises `PixieLookupError` rather than
        guessing when a query matches several targets.
        """
        target: Union[str, PixieTarget] = self.hook(PixieEvent.LookupTarget, target)
        if isinstance(target, str):
            _target = self._match_target(target)
            if _target is not None:
                target = _target
        if isinstance(target, PixieTarget):
            target.resolve_ip()

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

    def lookup_image(
        self, name: str, target: "PixieTarget | None" = None
    ) -> "PixieImage | None":
        """The best-matching image, or `None` when none matches.

        `match` may return a bool or a comparable score; anything falsy --
        `False`, `None`, `0`, an empty `re.Match`-less result -- means "not
        this one". Scores are compared against each other, so a `match`
        returning a non-comparable type raises rather than silently ordering
        by chance.
        """
        best_score, best_image = None, None
        for img_name, image in self.images.items():
            score = image.match(img_name, name)
            if not score:
                continue
            if best_score is None or score > best_score:
                best_score, best_image = score, image
        return self.hook(PixieEvent.FoundTargetImage, best_image, target=target)

    def lookup_dhcpzone(self, name: str, target: PixieTarget = None) -> "DHCPZone|None":
        """The zone for `name`, or the most specific one containing the target's IP.

        The resolved id is cached back onto `target.dhcpzone`.
        """
        if not name and target:
            if target.dhcpzone:
                name = target.dhcpzone
            elif target.resolve_ip().ip:
                # Containment needs a real address: a MAC-keyed target whose
                # hostname never resolved has `ip == ""`, and `"" in network`
                # raises AttributeError rather than returning False.
                # The most specific containing zone wins: with 10.0.0.0/16 and
                # 10.0.5.0/24 configured, a host in the /24 belongs to the /24
                # whichever order the zones happen to be declared in.
                containing = [
                    (zone.network.prefixlen, zone_id)
                    for zone_id, zone in self.dhcpzones.items()
                    if zone.network and target.ip in zone.network
                ]
                if containing:
                    name = max(containing)[1]
                    target.dhcpzone = name
        zone = self.dhcpzones.get(name, None)
        return self.hook(PixieEvent.FoundTargetDhcpzone, zone, target=target)

    def make_context(
        self,
        target: "PixieTarget",
        globals: list[dict] = None,
        require_image: bool = True,
    ) -> PixieContext:
        """Build the render context for `target`.

        Resolves the image and zone, layers the globals (engine, then image, zone
        and target), and attaches a renderer over the configured template roots.
        `require_image=False` tolerates an image that is no longer configured, so
        cleanup can still run.
        """
        target.resolve_ip()
        image = self.lookup_image(target.image, target)
        if not image and not require_image:
            # `complete` only needs the zone, to disarm DHCP. Refusing to clean
            # up a machine because its image was retired from the config leaves
            # it armed forever.
            LOGGER.warning(
                "target %s: image %r is not configured; continuing without it",
                target._id,
                target.image,
            )
            image = PixieImage(_id=target.image or "")
        elif not image:
            known = ", ".join(sorted(str(i) for i in self.images)) or "none configured"
            raise PixieLookupError(
                f"target {target._id!r} wants image {target.image!r}, which no "
                f"configured image matches (images: {known})"
            )
        LOGGER.info(f"Found target image: {target.image}")

        zone = self.lookup_dhcpzone(target.dhcpzone, target)
        if not zone:
            known = (
                ", ".join(sorted(str(z) for z in self.dhcpzones)) or "none configured"
            )
            wanted = target.dhcpzone or (
                f"a zone containing {target.ip}"
                if target.ip
                else "a zone, but the target has neither a dhcpzone nor an ip"
            )
            raise PixieLookupError(
                f"target {target._id!r} needs {wanted} (zones: {known})"
            )
        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)
        _globals = [self._without_reserved_keys(g) for g in _globals]

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

        ctx: PixieContext
        undefined = self.templates_undefined
        if undefined not in UNDEFINED_MODES:
            raise PixieConfigError(
                f"templates_undefined must be one of {', '.join(UNDEFINED_MODES)}, "
                f"not {undefined!r}"
            )
        declared = self._config.get("templates") or []
        if not isinstance(declared, (list, tuple)):
            declared = [declared]  # `templates: dir` is one root, not one per letter
        roots = [
            root if isinstance(root, pathapi.Path) else parse_path(str(root))
            for root in declared
        ]
        ctx._loader = tplx.Loader(
            roots,
            undefined=undefined,
            globals={
                "shell_quote": shell_quote,
                "Path": pathapi.Path,
                "Uri": URIPath,
            },
            # `_id` names an image or a target, and templates use it.
            encoder=tplx.DataEncoder(expose=["_id"]),
        )
        ctx.version = f"netboot-v{self.VERSION}"
        # `netboot.dhcp` fires BuildDhcpOptions through the engine while building
        # a reservation, and it only has the context.
        ctx._netboot_ = self
        return self.hook(PixieEvent.PixieContextForTarget, ctx, target=target)

    def _without_reserved_keys(self, values: dict) -> dict:
        """Drop globals that would overwrite the context's own fields.

        A global named `target` or `version` would replace the context's own
        value, and one named `render` or `searchpaths` its method: silently,
        and only for the targets whose globals happen to carry that name.
        """
        if not isinstance(values, Mapping):
            return values
        cls = self._ctxcls
        reserved = _config_keys(cls) | {n for n in dir(cls) if not n.startswith("_")}
        clashes = reserved.intersection(values)
        if not clashes:
            return values
        LOGGER.warning(
            "ignoring global(s) %s: those names are the render context's own fields",
            ", ".join(sorted(clashes)),
        )
        return {k: v for k, v in values.items() if k not in clashes}

    def initialize(self, target: "PixieTarget", **options: object) -> PixieContext:
        """Arm DHCP for `target` and return its context.

        A backend that fails is a warning; it raises only if none could be armed.
        `options` (`initiate`'s `iscsi=`) reach the Start/EndPixieInitialize
        hooks as their keyword arguments.
        """
        target = self.hook(PixieEvent.StartPixieInitialize, target, **options)
        ctx = self.make_context(target)
        ctx = ctx.pxe_init(self)
        return self.hook(PixieEvent.EndPixieInitialize, ctx, **options)

    def complete(self, target: "PixieTarget", **options: object) -> PixieContext:
        """Disarm DHCP for `target` and return its context.

        Every backend is tried; one that fails is logged as a warning.
        `options` reach the Start/EndPixieComplete hooks as keyword arguments.
        """
        target = self.hook(PixieEvent.StartPixieComplete, target, **options)
        ctx = self.make_context(target, require_image=False)
        ctx = ctx.pxe_complete(self)
        return self.hook(PixieEvent.EndPixieComplete, ctx, **options)

targets instance-attribute

targets: dict[str, PixieTarget]

dhcpzones instance-attribute

dhcpzones: dict[str, DHCPZone]

images instance-attribute

images: dict[str, PixieImage]

repos instance-attribute

repos: dict[str, Repository]

globals instance-attribute

globals: dict[str, object]

templates_undefined class-attribute instance-attribute

templates_undefined: str = 'strict'

VERSION class-attribute instance-attribute

VERSION = __version__

hook

hook(
    event: PixieEvent,
    __value: T = None,
    /,
    **kwargs: object,
) -> T

Run the hook chain for event, threading value through it.

Each hook is called as f(event, netboot, value, kwargs) and must return the value to pass on. value is positional-only.

Source code in src/netboot/engine.py

def hook(
    self: "Pixie|_ty.Sequence[_PixieHook]",
    event: PixieEvent,
    __value: T = None,
    /,
    **kwargs: object,
) -> T:
    """Run the hook chain for `event`, threading `value` through it.

    Each hook is called as `f(event, netboot, value, kwargs)` and **must
    return** the value to pass on. `value` is positional-only.
    """
    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

lookup_target

lookup_target(target: str) -> PixieTarget | None

Find the target a user named, or None.

Exact id first, then an exact hostname/MAC/IP match anywhere in the table, then a unique hostname prefix. Raises PixieLookupError rather than guessing when a query matches several targets.

Source code in src/netboot/engine.py

def lookup_target(self, target: str) -> "PixieTarget|None":
    """Find the target a user named, or `None`.

    Exact id first, then an exact hostname/MAC/IP match anywhere in the table,
    then a unique hostname prefix. Raises `PixieLookupError` rather than
    guessing when a query matches several targets.
    """
    target: Union[str, PixieTarget] = self.hook(PixieEvent.LookupTarget, target)
    if isinstance(target, str):
        _target = self._match_target(target)
        if _target is not None:
            target = _target
    if isinstance(target, PixieTarget):
        target.resolve_ip()

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

lookup_image

lookup_image(
    name: str, target: PixieTarget | None = None
) -> PixieImage | None

The best-matching image, or None when none matches.

match may return a bool or a comparable score; anything falsy -- False, None, 0, an empty re.Match-less result -- means "not this one". Scores are compared against each other, so a match returning a non-comparable type raises rather than silently ordering by chance.

Source code in src/netboot/engine.py

def lookup_image(
    self, name: str, target: "PixieTarget | None" = None
) -> "PixieImage | None":
    """The best-matching image, or `None` when none matches.

    `match` may return a bool or a comparable score; anything falsy --
    `False`, `None`, `0`, an empty `re.Match`-less result -- means "not
    this one". Scores are compared against each other, so a `match`
    returning a non-comparable type raises rather than silently ordering
    by chance.
    """
    best_score, best_image = None, None
    for img_name, image in self.images.items():
        score = image.match(img_name, name)
        if not score:
            continue
        if best_score is None or score > best_score:
            best_score, best_image = score, image
    return self.hook(PixieEvent.FoundTargetImage, best_image, target=target)

lookup_dhcpzone

lookup_dhcpzone(
    name: str, target: PixieTarget = None
) -> DHCPZone | None

The zone for name, or the most specific one containing the target's IP.

The resolved id is cached back onto target.dhcpzone.

Source code in src/netboot/engine.py

def lookup_dhcpzone(self, name: str, target: PixieTarget = None) -> "DHCPZone|None":
    """The zone for `name`, or the most specific one containing the target's IP.

    The resolved id is cached back onto `target.dhcpzone`.
    """
    if not name and target:
        if target.dhcpzone:
            name = target.dhcpzone
        elif target.resolve_ip().ip:
            # Containment needs a real address: a MAC-keyed target whose
            # hostname never resolved has `ip == ""`, and `"" in network`
            # raises AttributeError rather than returning False.
            # The most specific containing zone wins: with 10.0.0.0/16 and
            # 10.0.5.0/24 configured, a host in the /24 belongs to the /24
            # whichever order the zones happen to be declared in.
            containing = [
                (zone.network.prefixlen, zone_id)
                for zone_id, zone in self.dhcpzones.items()
                if zone.network and target.ip in zone.network
            ]
            if containing:
                name = max(containing)[1]
                target.dhcpzone = name
    zone = self.dhcpzones.get(name, None)
    return self.hook(PixieEvent.FoundTargetDhcpzone, zone, target=target)

make_context

make_context(
    target: PixieTarget,
    globals: list[dict] = None,
    require_image: bool = True,
) -> PixieContext

Build the render context for target.

Resolves the image and zone, layers the globals (engine, then image, zone and target), and attaches a renderer over the configured template roots. require_image=False tolerates an image that is no longer configured, so cleanup can still run.

Source code in src/netboot/engine.py

def make_context(
    self,
    target: "PixieTarget",
    globals: list[dict] = None,
    require_image: bool = True,
) -> PixieContext:
    """Build the render context for `target`.

    Resolves the image and zone, layers the globals (engine, then image, zone
    and target), and attaches a renderer over the configured template roots.
    `require_image=False` tolerates an image that is no longer configured, so
    cleanup can still run.
    """
    target.resolve_ip()
    image = self.lookup_image(target.image, target)
    if not image and not require_image:
        # `complete` only needs the zone, to disarm DHCP. Refusing to clean
        # up a machine because its image was retired from the config leaves
        # it armed forever.
        LOGGER.warning(
            "target %s: image %r is not configured; continuing without it",
            target._id,
            target.image,
        )
        image = PixieImage(_id=target.image or "")
    elif not image:
        known = ", ".join(sorted(str(i) for i in self.images)) or "none configured"
        raise PixieLookupError(
            f"target {target._id!r} wants image {target.image!r}, which no "
            f"configured image matches (images: {known})"
        )
    LOGGER.info(f"Found target image: {target.image}")

    zone = self.lookup_dhcpzone(target.dhcpzone, target)
    if not zone:
        known = (
            ", ".join(sorted(str(z) for z in self.dhcpzones)) or "none configured"
        )
        wanted = target.dhcpzone or (
            f"a zone containing {target.ip}"
            if target.ip
            else "a zone, but the target has neither a dhcpzone nor an ip"
        )
        raise PixieLookupError(
            f"target {target._id!r} needs {wanted} (zones: {known})"
        )
    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)
    _globals = [self._without_reserved_keys(g) for g in _globals]

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

    ctx: PixieContext
    undefined = self.templates_undefined
    if undefined not in UNDEFINED_MODES:
        raise PixieConfigError(
            f"templates_undefined must be one of {', '.join(UNDEFINED_MODES)}, "
            f"not {undefined!r}"
        )
    declared = self._config.get("templates") or []
    if not isinstance(declared, (list, tuple)):
        declared = [declared]  # `templates: dir` is one root, not one per letter
    roots = [
        root if isinstance(root, pathapi.Path) else parse_path(str(root))
        for root in declared
    ]
    ctx._loader = tplx.Loader(
        roots,
        undefined=undefined,
        globals={
            "shell_quote": shell_quote,
            "Path": pathapi.Path,
            "Uri": URIPath,
        },
        # `_id` names an image or a target, and templates use it.
        encoder=tplx.DataEncoder(expose=["_id"]),
    )
    ctx.version = f"netboot-v{self.VERSION}"
    # `netboot.dhcp` fires BuildDhcpOptions through the engine while building
    # a reservation, and it only has the context.
    ctx._netboot_ = self
    return self.hook(PixieEvent.PixieContextForTarget, ctx, target=target)

initialize

initialize(
    target: PixieTarget, **options: object
) -> PixieContext

Arm DHCP for target and return its context.

A backend that fails is a warning; it raises only if none could be armed. options (initiate's iscsi=) reach the Start/EndPixieInitialize hooks as their keyword arguments.

Source code in src/netboot/engine.py

def initialize(self, target: "PixieTarget", **options: object) -> PixieContext:
    """Arm DHCP for `target` and return its context.

    A backend that fails is a warning; it raises only if none could be armed.
    `options` (`initiate`'s `iscsi=`) reach the Start/EndPixieInitialize
    hooks as their keyword arguments.
    """
    target = self.hook(PixieEvent.StartPixieInitialize, target, **options)
    ctx = self.make_context(target)
    ctx = ctx.pxe_init(self)
    return self.hook(PixieEvent.EndPixieInitialize, ctx, **options)

complete

complete(
    target: PixieTarget, **options: object
) -> PixieContext

Disarm DHCP for target and return its context.

Every backend is tried; one that fails is logged as a warning. options reach the Start/EndPixieComplete hooks as keyword arguments.

Source code in src/netboot/engine.py

def complete(self, target: "PixieTarget", **options: object) -> PixieContext:
    """Disarm DHCP for `target` and return its context.

    Every backend is tried; one that fails is logged as a warning.
    `options` reach the Start/EndPixieComplete hooks as keyword arguments.
    """
    target = self.hook(PixieEvent.StartPixieComplete, target, **options)
    ctx = self.make_context(target, require_image=False)
    ctx = ctx.pxe_complete(self)
    return self.hook(PixieEvent.EndPixieComplete, ctx, **options)

Context

netboot.PixieContext

Bases: Namespace

Everything a template can see for one target.

Built by Pixie.make_context, never constructed directly: it carries the target, its image and zone, the repositories, and its own renderer.

Source code in src/netboot/_context.py

class PixieContext(Namespace):
    """Everything a template can see for one target.

    Built by `Pixie.make_context`, never constructed directly: it carries the
    target, its image and zone, the repositories, and its own renderer.
    """

    target: PixieTarget
    image: PixieImage
    dhcpzone: DHCPZone
    repos: dict[str, Repository]
    generated: datetime.datetime
    resources: dict[str, Resource]
    version: str
    _loader: tplx.Loader

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

    def resource(
        self, name: Union[str, Resource], service: "str | None" = None
    ) -> "URIPath | None":
        """Resolve a resource to a fetchable URI through its repository.

        Takes a resource id (looked up in `.resources`) or a `Resource` directly.
        `None` if the repo or the resource is unknown.
        """
        if isinstance(name, Resource):
            repo = self.repos.get(getattr(name, "src", None), None)
            path = getattr(name, "path", None)
            if not path:
                return
        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) -> "Repository | None":
        """The `Repository` backing resource `name`, or `None`."""
        resource = self.resources.get(name)
        if not resource:
            return
        return self.repos.get(resource.src)

    def pxe_init(self, config: "Pixie") -> "PixieContext":
        """Arm every DHCP backend for this target, succeeding if any one does.

        Every backend is tried. A backend that fails is logged as a warning and
        skipped, so one server being offline does not block the others. When
        none could be armed the first error is raised: as itself when netboot
        or dhcpctl raised it on purpose, else as a `PixieError` chained to it.
        """
        first_error = None
        armed = 0
        if not self.dhcpzone.dhcpservers:
            LOGGER.warning(
                "zone %s has no dhcpservers: nothing was armed for %s",
                self.dhcpzone._id,
                self.target._id,
            )
        for dhcpserver in self.dhcpzone.dhcpservers:
            try:
                arm(self, dhcpserver)
            except Exception as exc:
                LOGGER.warning(
                    "could not arm %s on %s: %s",
                    self.target._id,
                    dhcpserver,
                    exc,
                    exc_info=LOGGER.isEnabledFor(logging.DEBUG),
                )
                first_error = first_error or exc
            else:
                armed += 1
        if first_error is not None and not armed:
            if isinstance(first_error, (PixieError, dhcpctl.DHCPCtlError)):
                raise first_error
            raise PixieError(
                f"no DHCP server armed {self.target._id!r}: {first_error}"
            ) from first_error
        return self

    def pxe_complete(self, config: "Pixie") -> "PixieContext":
        """Disarm every DHCP backend; a failure is a warning, never an error.

        Every backend is tried, and one that fails is logged and skipped, so
        the rest are still disarmed.
        """
        for dhcpserver in self.dhcpzone.dhcpservers:
            try:
                # Removes the entry unless the config says to keep it
                # (`dhcp_complete:`).
                disarm(self, dhcpserver)
            except Exception as exc:
                LOGGER.warning(
                    "could not disarm %s on %s: %s",
                    self.target._id,
                    dhcpserver,
                    exc,
                    exc_info=LOGGER.isEnabledFor(logging.DEBUG),
                )
        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 unset values so they never yield a spurious name. The null MAC
        # is the same case as the unspecified IP: every MAC-less target would
        # otherwise share the candidate `00-00-00-00-00-00.<name>`, and one
        # stray file of that name would apply to all of them.
        # `is_wildcard` knows every spelling of "every local address", a
        # `%zone` suffix and `None` included; the literal tuple this replaced
        # knew two of them.
        ip_name = "" if netimps.is_wildcard(ip) else str(ip)
        mac = self.target.mac
        mac_name = mac.format("-") if str(mac) != PixieTarget._NULL_MAC else ""
        names = []
        for version in [
            mac_name,
            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]:
        """Template directories for this context: target first, then image.

        These extend the engine-wide template roots rather than replacing them; a
        relative entry is resolved inside each root.
        """
        return [*self.target.template_path, *self.image.template_path]

    def render(self, filename: str, strict: bool = True) -> "str | bytes | None":
        """Render `filename` against this context and return the result.

        The name is resolved by the template search (MAC, hostname, IP, then the
        bare name), through `tplx`, and so are the includes inside it. The
        template sees `ctx` and each of the context's fields (`target`, `image`,
        ...). Text, or **`bytes`** for a file the copy engine claims, which is
        every suffix no other engine owns -- it is a byte-exact copy, so it is
        never decoded. `strict=False` logs the failure at DEBUG and returns
        `None` instead of raising.
        """
        data = dict(vars(self))
        data["ctx"] = self
        paths = [
            path if isinstance(path, pathapi.Path) else parse_path(str(path))
            for path in self.searchpaths
        ]
        try:
            return self._loader.render(
                filename, data, candidates=self._template_names, paths=paths
            )
        except Exception as e:
            if strict:
                raise e
            else:
                LOGGER.debug(
                    f"While rendering [{filename}] encountered an exeption:\t{repr(e)}"
                )
                return None

target instance-attribute

target: PixieTarget

image instance-attribute

image: PixieImage

dhcpzone instance-attribute

dhcpzone: DHCPZone

repos instance-attribute

repos: dict[str, Repository]

generated instance-attribute

generated: datetime = datetime.datetime.now()

resources instance-attribute

resources: dict[str, Resource]

version instance-attribute

version: str

searchpaths property

searchpaths: list[Path]

Template directories for this context: target first, then image.

These extend the engine-wide template roots rather than replacing them; a relative entry is resolved inside each root.

resource

resource(
    name: Union[str, Resource], service: str | None = None
) -> URIPath | None

Resolve a resource to a fetchable URI through its repository.

Takes a resource id (looked up in .resources) or a Resource directly. None if the repo or the resource is unknown.

Source code in src/netboot/_context.py

def resource(
    self, name: Union[str, Resource], service: "str | None" = None
) -> "URIPath | None":
    """Resolve a resource to a fetchable URI through its repository.

    Takes a resource id (looked up in `.resources`) or a `Resource` directly.
    `None` if the repo or the resource is unknown.
    """
    if isinstance(name, Resource):
        repo = self.repos.get(getattr(name, "src", None), None)
        path = getattr(name, "path", None)
        if not path:
            return
    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)

resource_repo

resource_repo(name: str) -> Repository | None

The Repository backing resource name, or None.

Source code in src/netboot/_context.py

def resource_repo(self, name: str) -> "Repository | None":
    """The `Repository` backing resource `name`, or `None`."""
    resource = self.resources.get(name)
    if not resource:
        return
    return self.repos.get(resource.src)

pxe_init

pxe_init(config: Pixie) -> PixieContext

Arm every DHCP backend for this target, succeeding if any one does.

Every backend is tried. A backend that fails is logged as a warning and skipped, so one server being offline does not block the others. When none could be armed the first error is raised: as itself when netboot or dhcpctl raised it on purpose, else as a PixieError chained to it.

Source code in src/netboot/_context.py

def pxe_init(self, config: "Pixie") -> "PixieContext":
    """Arm every DHCP backend for this target, succeeding if any one does.

    Every backend is tried. A backend that fails is logged as a warning and
    skipped, so one server being offline does not block the others. When
    none could be armed the first error is raised: as itself when netboot
    or dhcpctl raised it on purpose, else as a `PixieError` chained to it.
    """
    first_error = None
    armed = 0
    if not self.dhcpzone.dhcpservers:
        LOGGER.warning(
            "zone %s has no dhcpservers: nothing was armed for %s",
            self.dhcpzone._id,
            self.target._id,
        )
    for dhcpserver in self.dhcpzone.dhcpservers:
        try:
            arm(self, dhcpserver)
        except Exception as exc:
            LOGGER.warning(
                "could not arm %s on %s: %s",
                self.target._id,
                dhcpserver,
                exc,
                exc_info=LOGGER.isEnabledFor(logging.DEBUG),
            )
            first_error = first_error or exc
        else:
            armed += 1
    if first_error is not None and not armed:
        if isinstance(first_error, (PixieError, dhcpctl.DHCPCtlError)):
            raise first_error
        raise PixieError(
            f"no DHCP server armed {self.target._id!r}: {first_error}"
        ) from first_error
    return self

pxe_complete

pxe_complete(config: Pixie) -> PixieContext

Disarm every DHCP backend; a failure is a warning, never an error.

Every backend is tried, and one that fails is logged and skipped, so the rest are still disarmed.

Source code in src/netboot/_context.py

def pxe_complete(self, config: "Pixie") -> "PixieContext":
    """Disarm every DHCP backend; a failure is a warning, never an error.

    Every backend is tried, and one that fails is logged and skipped, so
    the rest are still disarmed.
    """
    for dhcpserver in self.dhcpzone.dhcpservers:
        try:
            # Removes the entry unless the config says to keep it
            # (`dhcp_complete:`).
            disarm(self, dhcpserver)
        except Exception as exc:
            LOGGER.warning(
                "could not disarm %s on %s: %s",
                self.target._id,
                dhcpserver,
                exc,
                exc_info=LOGGER.isEnabledFor(logging.DEBUG),
            )
    return self

render

render(
    filename: str, strict: bool = True
) -> str | bytes | None

Render filename against this context and return the result.

The name is resolved by the template search (MAC, hostname, IP, then the bare name), through tplx, and so are the includes inside it. The template sees ctx and each of the context's fields (target, image, ...). Text, or bytes for a file the copy engine claims, which is every suffix no other engine owns -- it is a byte-exact copy, so it is never decoded. strict=False logs the failure at DEBUG and returns None instead of raising.

Source code in src/netboot/_context.py

def render(self, filename: str, strict: bool = True) -> "str | bytes | None":
    """Render `filename` against this context and return the result.

    The name is resolved by the template search (MAC, hostname, IP, then the
    bare name), through `tplx`, and so are the includes inside it. The
    template sees `ctx` and each of the context's fields (`target`, `image`,
    ...). Text, or **`bytes`** for a file the copy engine claims, which is
    every suffix no other engine owns -- it is a byte-exact copy, so it is
    never decoded. `strict=False` logs the failure at DEBUG and returns
    `None` instead of raising.
    """
    data = dict(vars(self))
    data["ctx"] = self
    paths = [
        path if isinstance(path, pathapi.Path) else parse_path(str(path))
        for path in self.searchpaths
    ]
    try:
        return self._loader.render(
            filename, data, candidates=self._template_names, paths=paths
        )
    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

One machine to provision, as configured under targets:.

Gaps are filled at construction: a MAC-shaped id becomes the MAC, an IP-shaped id the IP, anything else the hostname. An unknown IP is looked up only by resolve_ip, which the engine calls for a target it acts on, so building the engine does no DNS.

Source code in src/netboot/_objects.py

class PixieTarget(Namespace, OpaqueMerge):
    """One machine to provision, as configured under `targets:`.

    Gaps are filled at construction: a MAC-shaped id becomes the MAC, an
    IP-shaped id the IP, anything else the hostname. An unknown IP is looked
    up only by `resolve_ip`, which the engine calls for a target it acts on,
    so building the engine does no DNS.
    """

    _id: str
    hostname: str
    #: An `IPAddress` once known, but `""` while it is not: a MAC-keyed target
    #: whose hostname did not resolve has no address, and callers test it for
    #: truthiness rather than assuming one.
    ip: "Union[IPAddress, str]"
    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"
    #: Config keys read without an annotation: the DHCP layers.
    _CONFIG_KEYS = frozenset({"dhcp_options", "dhcp_when", "dhcp_complete"})

    def __init__(self, **kwargs: object) -> None:
        for prop, key in {
            "dhcpzone": "",
            "mac": "",
            "ip": "",
            "image": "",
            "hostname": "",
            "template_path": [],
        }.items():
            kwargs.setdefault(prop, key)
        super().__init__(**kwargs)
        if isinstance(self.mac, (int, float)):
            # YAML 1.1 reads an unquoted all-digit MAC as a base-60 number,
            # which would parse as some other machine's MAC.
            raise PixieConfigError(
                f"target {self._id!r}: mac {self.mac!r} is a number, not a MAC; "
                "quote it (YAML reads an unquoted 52:54:00:12:34:56 as base 60)"
            )
        # 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.is_valid(str(self._id)):
                self.mac = self._id
            else:
                self.mac = self._NULL_MAC
        if not isinstance(self.mac, MACAddress):
            self.mac = MACAddress(self.mac)
        id_str = str(getattr(self, "_id", "") or "")
        id_is_mac = MACAddress.is_valid(id_str)
        id_is_ip = netimps.is_valid(id_str, IPAddress)
        if not self.ip and id_is_ip:
            self.ip = id_str
        if not self.hostname and not id_is_mac and not id_is_ip:
            self.hostname = id_str
        self.ip = netimps.parse(self.ip, IPAddress) if self.ip else ""
        self.hostname = self.hostname.lower()
        self._ip_looked_up = False

    def resolve_ip(self) -> "PixieTarget":
        """Fill an unknown `ip` from `hostname`: one forward lookup, ever.

        Never retried: a name that does not resolve answers the same next
        time, so it leaves `ip` unset with a warning, and never raises.
        """
        if self.ip or not self.hostname or self._ip_looked_up:
            return self
        self._ip_looked_up = True
        try:
            resolved = netimps.resolve(self.hostname)
        except ValueError as exc:  # malformed name: not worth killing the run
            LOGGER.warning(
                "target %s: %r is not resolvable: %s", self._id, self.hostname, exc
            )
            return self
        if resolved:
            self.ip = netimps.parse(resolved[0], IPAddress)
        else:
            LOGGER.warning(
                "target %s: hostname %r did not resolve; its ip stays unset",
                self._id,
                self.hostname,
            )
        return self

hostname instance-attribute

hostname: str = self.hostname.lower()

ip instance-attribute

ip: Union[IPAddress, str] = (
    netimps.parse(self.ip, IPAddress) if self.ip else ""
)

mac instance-attribute

mac: MACAddress

image instance-attribute

image: str

dhcpzone instance-attribute

dhcpzone: str

globals instance-attribute

globals: dict

template_path instance-attribute

template_path: list[Union[str, Path]]

resolve_ip

resolve_ip() -> PixieTarget

Fill an unknown ip from hostname: one forward lookup, ever.

Never retried: a name that does not resolve answers the same next time, so it leaves ip unset with a warning, and never raises.

Source code in src/netboot/_objects.py

def resolve_ip(self) -> "PixieTarget":
    """Fill an unknown `ip` from `hostname`: one forward lookup, ever.

    Never retried: a name that does not resolve answers the same next
    time, so it leaves `ip` unset with a warning, and never raises.
    """
    if self.ip or not self.hostname or self._ip_looked_up:
        return self
    self._ip_looked_up = True
    try:
        resolved = netimps.resolve(self.hostname)
    except ValueError as exc:  # malformed name: not worth killing the run
        LOGGER.warning(
            "target %s: %r is not resolvable: %s", self._id, self.hostname, exc
        )
        return self
    if resolved:
        self.ip = netimps.parse(resolved[0], IPAddress)
    else:
        LOGGER.warning(
            "target %s: hostname %r did not resolve; its ip stays unset",
            self._id,
            self.hostname,
        )
    return self

netboot.PixieImage

Bases: Resource

What a target boots, as configured under images:.

Also a content.Resource, so it carries src/path addressing its artifacts in a repository. Those play no part in template lookup.

Source code in src/netboot/_objects.py

class PixieImage(Resource):
    """What a target boots, as configured under `images:`.

    Also a `content.Resource`, so it carries `src`/`path` addressing its
    artifacts in a repository. Those play no part in template lookup.
    """

    template_path: list["Path"]
    globals: dict
    _CONFIG_KEYS = PixieTarget._CONFIG_KEYS

    def __init__(self, **kwargs: object) -> None:
        # `searchpaths` reads this on every render, so an image that simply
        # does not declare one must still have the empty list.
        kwargs.setdefault("template_path", [])
        super().__init__(**kwargs)

    def match(self, name: str, check: str) -> object:
        """Does image `name` serve `check`? Override point for image selection.

        Return a bool, or a comparable score so `lookup_image` can prefer the best
        of several matches. Anything falsy means "not this one".
        """
        return name == check

template_path instance-attribute

template_path: list[Path]

globals instance-attribute

globals: dict

match

match(name: str, check: str) -> object

Does image name serve check? Override point for image selection.

Return a bool, or a comparable score so lookup_image can prefer the best of several matches. Anything falsy means "not this one".

Source code in src/netboot/_objects.py

def match(self, name: str, check: str) -> object:
    """Does image `name` serve `check`? Override point for image selection.

    Return a bool, or a comparable score so `lookup_image` can prefer the best
    of several matches. Anything falsy means "not this one".
    """
    return name == check

Events

netboot.PixieEvent

Bases: StrEnum

The points in the engine where hooks run.

The string value carries the PixieEvent. prefix (PixieEvent.LookupTarget), which hooks comparing against raw strings rely on.

Source code in src/netboot/engine.py

class PixieEvent(StrEnum):
    """The points in the engine where hooks run.

    The string value carries the `PixieEvent.` prefix (`PixieEvent.LookupTarget`),
    which hooks comparing against raw strings rely on.
    """

    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"
    BuildDhcpOptions = "PixieEvent.BuildDhcpOptions"
    StartPixieInitialize = "PixieEvent.StartPixieInitialize"
    EndPixieInitialize = "PixieEvent.EndPixieInitialize"
    StartPixieComplete = "PixieEvent.StartPixieComplete"
    EndPixieComplete = "PixieEvent.EndPixieComplete"

NewPixieObject class-attribute instance-attribute

NewPixieObject = 'PixieEvent.NewPixieObject'

StartPixieInit class-attribute instance-attribute

StartPixieInit = 'PixieEvent.StartPixieInit'

SetPixieProperty class-attribute instance-attribute

SetPixieProperty = 'PixieEvent.SetPixieProperty'

PixieInitiated class-attribute instance-attribute

PixieInitiated = 'PixieEvent.PixieInitiated'

LookupTarget class-attribute instance-attribute

LookupTarget = 'PixieEvent.LookupTarget'

FoundTarget class-attribute instance-attribute

FoundTarget = 'PixieEvent.FoundTarget'

FoundTargetImage class-attribute instance-attribute

FoundTargetImage = 'PixieEvent.FoundTargetImage'

FoundTargetDhcpzone class-attribute instance-attribute

FoundTargetDhcpzone = 'PixieEvent.FoundTargetDhcpzone'

PixieContextForTarget class-attribute instance-attribute

PixieContextForTarget = 'PixieEvent.PixieContextForTarget'

BuildDhcpOptions class-attribute instance-attribute

BuildDhcpOptions = 'PixieEvent.BuildDhcpOptions'

StartPixieInitialize class-attribute instance-attribute

StartPixieInitialize = 'PixieEvent.StartPixieInitialize'

EndPixieInitialize class-attribute instance-attribute

EndPixieInitialize = 'PixieEvent.EndPixieInitialize'

StartPixieComplete class-attribute instance-attribute

StartPixieComplete = 'PixieEvent.StartPixieComplete'

EndPixieComplete class-attribute instance-attribute

EndPixieComplete = 'PixieEvent.EndPixieComplete'

DHCP

The servers are dhcpctl's; this module builds each target's reservation for them.

netboot.dhcp

DHCP zones, and what a target's reservation is on each of their servers.

The servers are dhcpctl's: a zone's dhcpservers are URIs, built with dhcpctl.connect, and a backend for a new scheme is registered with dhcpctl.register_backend (or the dhcpctl.backends entry point). What stays here is what is about targets: the zone -> image -> target layering of dhcp_options, dhcp_when and dhcp_complete, the BuildDhcpOptions hook, and each server's own secrets from its dhcpservers entry.

OWNER module-attribute

OWNER = 'netboot'

LEGACY_OPTION_NAMES module-attribute

LEGACY_OPTION_NAMES = {
    "domain-name-servers": "dns",
    "boot-file-name": "file",
    "tftp-server-name": "tftp-server",
    "host-name": "hostname",
    "lease-time": "ip-address-lease-time",
}

NETBOOT_CODES module-attribute

NETBOOT_CODES = dhcpctl.option_codes(LEGACY_OPTION_NAMES)

DHCPZone

Bases: Namespace, OpaqueMerge

A network netboot can provision into, as configured under dhcpzones:.

Derives network from a CIDR gateway, coerces nameservers/search to lists, and builds dhcpservers URIs into dhcpctl servers.

Source code in src/netboot/dhcp.py

class DHCPZone(Namespace, OpaqueMerge):
    """A network netboot can provision into, as configured under `dhcpzones:`.

    Derives `network` from a CIDR `gateway`, coerces `nameservers`/`search` to
    lists, and builds `dhcpservers` URIs into dhcpctl servers.
    """

    network: IPNetwork
    gateway: "_ty.Optional[IPAddress]"
    domain: "_ty.Optional[str]"
    search: "list[str]"
    nameservers: "list[IPAddress]"
    globals: dict
    dhcpservers: "list[dhcpctl.DHCPServer]"
    #: Config keys read without an annotation.
    _CONFIG_KEYS = frozenset(
        {"netmask", "subnet_id", "scope", "dhcp_when", "dhcp_complete"}
    )

    @property
    def nameserver(self) -> "IPAddress | str":
        """The first nameserver, or `""` when none is configured."""
        return self.nameservers[0] if self.nameservers else ""

    def get_local_server(
        self, servers: "list[IPAddress]", default: IPAddress
    ) -> IPAddress:
        """The first server that lives inside this zone's network, else `default`.

        Accepts the strings config and globals actually hold, not just parsed
        addresses: `"10.0.0.5" in network` would otherwise raise.
        """
        if isinstance(servers, (str, bytes)) or not isinstance(servers, (list, tuple)):
            servers = [servers]
        for server in servers:
            address = netimps.try_parse(server, IPAddress)
            if address is None:
                LOGGER.warning("ignoring unparseable server address %r", server)
                continue
            if self.network and address in self.network:
                return address
        return default

    def __init__(self, **kwargs: object) -> None:
        _gateway = kwargs.get("gateway")
        _network = kwargs.get("network")
        _netmask = kwargs.get("netmask")
        if _gateway and not _network:
            gw: netimps.IPv4Interface = netimps.parse(_gateway, IPInterface)
            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 = netimps.parse(_network, IPNetwork)
            _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)
        codes = kwargs.pop("codes", None) or NETBOOT_CODES
        kwargs.setdefault("dhcpservers", [])
        super().__init__(**kwargs)
        self.dhcpservers = [
            server_from(server, codes) for server in (self.dhcpservers or [])
        ]
        network = getattr(self, "network", None)
        self.network = netimps.parse(network, IPNetwork) if network else None
        for prop, ctr in [
            ("nameservers", lambda v: netimps.try_parse(v, IPAddress)),
            ("search", str),
        ]:
            value = getattr(self, prop, None)
            if not value:
                setattr(self, prop, [])
                continue
            # Copy rather than normalise in place: the list may be the caller's
            # config object, shared with other zones through a merge.
            values = list(value) if isinstance(value, (list, tuple)) else [value]
            converted = []
            for val in values:
                new = ctr(val)
                if new is None:
                    # try_parse says "not an address"; dropping it silently
                    # would hand DHCP clients a zone with fewer nameservers
                    # than the operator wrote.
                    LOGGER.warning(
                        "zone %s: ignoring %s entry %r, which is not an address",
                        kwargs.get("_id", "<unnamed>"),
                        prop,
                        val,
                    )
                    continue
                converted.append(new)
            setattr(self, prop, converted)

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

        if self.gateway:
            self.gateway = netimps.parse(self.gateway, IPAddress)

gateway instance-attribute

gateway: '_ty.Optional[IPAddress]'

domain instance-attribute

domain: '_ty.Optional[str]'

search instance-attribute

search: 'list[str]'

nameservers instance-attribute

nameservers: 'list[IPAddress]'

globals instance-attribute

globals: dict

nameserver property

nameserver: 'IPAddress | str'

The first nameserver, or "" when none is configured.

dhcpservers instance-attribute

dhcpservers: "list[dhcpctl.DHCPServer]" = [
    server_from(server, codes)
    for server in self.dhcpservers or []
]

network instance-attribute

network: IPNetwork = (
    netimps.parse(network, IPNetwork) if network else None
)

get_local_server

get_local_server(
    servers: "list[IPAddress]", default: IPAddress
) -> IPAddress

The first server that lives inside this zone's network, else default.

Accepts the strings config and globals actually hold, not just parsed addresses: "10.0.0.5" in network would otherwise raise.

Source code in src/netboot/dhcp.py

def get_local_server(
    self, servers: "list[IPAddress]", default: IPAddress
) -> IPAddress:
    """The first server that lives inside this zone's network, else `default`.

    Accepts the strings config and globals actually hold, not just parsed
    addresses: `"10.0.0.5" in network` would otherwise raise.
    """
    if isinstance(servers, (str, bytes)) or not isinstance(servers, (list, tuple)):
        servers = [servers]
    for server in servers:
        address = netimps.try_parse(server, IPAddress)
        if address is None:
            LOGGER.warning("ignoring unparseable server address %r", server)
            continue
        if self.network and address in self.network:
            return address
    return default

codes_for

codes_for(
    dhcp: "dict|None" = None,
) -> "type[dhcpctl.OptionCodes]"

The codemap a config's dhcp: section declares, over netboot's.

dhcp.options maps a name to {code: <n>, type: <pydhcp codec>} or to a code or another name (see dhcpctl.option_codes).

:raises PixieConfigError: the section cannot be used.

Source code in src/netboot/dhcp.py

def codes_for(dhcp: "dict|None" = None) -> "type[dhcpctl.OptionCodes]":
    """The codemap a config's `dhcp:` section declares, over netboot's.

    `dhcp.options` maps a name to `{code: <n>, type: <pydhcp codec>}` or to a
    code or another name (see `dhcpctl.option_codes`).

    :raises PixieConfigError: the section cannot be used.
    """

    dhcp = dhcp or {}
    if not isinstance(dhcp, dict) or not isinstance(dhcp.get("options") or {}, dict):
        raise PixieConfigError("dhcp.options must be a mapping of option names")
    declared = dhcp.get("options")
    if not declared:
        return NETBOOT_CODES
    try:
        return dhcpctl.option_codes(declared, base=NETBOOT_CODES)
    except dhcpctl.ConfigError as exc:
        raise PixieConfigError(f"dhcp.{exc}") from exc

connect

connect(
    uri: str,
    codes: "type[dhcpctl.OptionCodes]|None" = None,
    secrets: "_abc.Mapping[str, str]|None" = None,
) -> dhcpctl.DHCPServer

The dhcpctl server for uri, with the secrets its config entry gives.

A secret given here overrides the one the URI carries (its user or password). Option names are read with codes (netboot's by default).

:raises PixieConfigError: the URI cannot be used.

Source code in src/netboot/dhcp.py

def connect(
    uri: str,
    codes: "type[dhcpctl.OptionCodes]|None" = None,
    secrets: "_abc.Mapping[str, str]|None" = None,
) -> dhcpctl.DHCPServer:
    """The dhcpctl server for `uri`, with the `secrets` its config entry gives.

    A secret given here overrides the one the URI carries (its user or
    password). Option names are read with `codes` (netboot's by default).

    :raises PixieConfigError: the URI cannot be used.
    """

    try:
        return dhcpctl.connect(
            uri, secrets=dict(secrets or {}), owner=OWNER, codes=codes or NETBOOT_CODES
        )
    except dhcpctl.ConfigError as exc:
        raise PixieConfigError(str(exc)) from exc

server_from

server_from(
    entry: "str | _abc.Mapping[str, object] | dhcpctl.DHCPServer",
    codes: "type[dhcpctl.OptionCodes] | None" = None,
) -> dhcpctl.DHCPServer

A dhcpservers entry as a server: a URI, or a mapping of uri: and the secrets that backend takes (windhcp and kea user/password, dhcpd keyname/omapi_key), each written in the config or produced by it (password: "{{ env.WINDHCP_PASSWORD }}").

:raises PixieConfigError: the entry cannot be used.

Source code in src/netboot/dhcp.py

def server_from(
    entry: "str | _abc.Mapping[str, object] | dhcpctl.DHCPServer",
    codes: "type[dhcpctl.OptionCodes] | None" = None,
) -> dhcpctl.DHCPServer:
    """A `dhcpservers` entry as a server: a URI, or a mapping of `uri:` and the
    secrets that backend takes (windhcp and kea `user`/`password`, dhcpd
    `keyname`/`omapi_key`), each written in the config or produced by it
    (`password: "{{ env.WINDHCP_PASSWORD }}"`).

    :raises PixieConfigError: the entry cannot be used.
    """

    if isinstance(entry, dhcpctl.DHCPServer):
        return entry
    if isinstance(entry, str):
        return connect(entry, codes)
    if not isinstance(entry, _abc.Mapping) or not isinstance(entry.get("uri"), str):
        raise PixieConfigError(
            "a dhcpservers entry is a URI, or a mapping with `uri:` and that "
            f"server's secrets, not {type(entry).__name__}"
        )
    uri = entry["uri"]
    secrets = {key: value for key, value in entry.items() if key != "uri"}
    try:
        backend = dhcpctl.get_backend(uri.partition(":")[0])
    except dhcpctl.ConfigError as exc:
        raise PixieConfigError(str(exc)) from exc
    accepted = sorted(set(backend.URI_SECRETS.values()))
    unknown = sorted(set(map(str, secrets)) - set(accepted))
    if unknown:
        raise PixieConfigError(
            f"dhcpservers entry {uri.partition('?')[0]!r}: unknown key(s) "
            f"{', '.join(unknown)}; it takes uri and {', '.join(accepted)}"
        )
    return connect(
        uri,
        codes,
        {key: str(value) for key, value in secrets.items() if value is not None},
    )

scope_for

scope_for(zone: DHCPZone) -> dhcpctl.Scope

A zone as the scope a reservation lives in: subnet_id/scope if set.

Source code in src/netboot/dhcp.py

def scope_for(zone: DHCPZone) -> dhcpctl.Scope:
    """A zone as the scope a reservation lives in: `subnet_id`/`scope` if set."""
    subnet_id = getattr(zone, "subnet_id", None)
    return dhcpctl.Scope(
        name=str(getattr(zone, "_id", "") or ""),
        network=getattr(zone, "network", None),
        subnet_id=int(subnet_id) if subnet_id is not None else None,
        scope_id=str(getattr(zone, "scope", "") or "") or None,
    )

zone_defaults

zone_defaults(
    zone: DHCPZone,
    codes: "type[dhcpctl.OptionCodes] | None" = None,
) -> dhcpctl.DHCPOptions

What the zone already knows, before anyone configures an option.

The bag reads its options with codes (netboot's by default), the codemap every layer merged onto it keeps.

Source code in src/netboot/dhcp.py

def zone_defaults(
    zone: DHCPZone, codes: "type[dhcpctl.OptionCodes] | None" = None
) -> dhcpctl.DHCPOptions:
    """What the zone already knows, before anyone configures an option.

    The bag reads its options with `codes` (netboot's by default), the codemap
    every layer merged onto it keeps.
    """
    known = {}
    if getattr(zone, "gateway", None):
        known["router"] = zone.gateway
    if getattr(zone, "nameservers", None):
        known["dns"] = list(zone.nameservers)
    if getattr(zone, "domain", None):
        known["domain-name"] = zone.domain
    if getattr(zone, "search", None):
        known["domain-search"] = list(zone.search)
    network = getattr(zone, "network", None)
    if network is not None and getattr(network, "netmask", None) is not None:
        known["subnet-mask"] = network.netmask
        broadcast = getattr(network, "broadcast_address", None)
        if broadcast is not None:
            known["broadcast-address"] = broadcast
    return (codes or NETBOOT_CODES).options(
        known, str(getattr(zone, "_id", "") or "zone")
    )

build_options

build_options(
    ctx: PixieContext, server: DHCPServer
) -> dhcpctl.DHCPOptions

The options for this target on this server, in merge order.

Later wins: zone defaults, the image's dhcp_options, the target's dhcp_options, then the BuildDhcpOptions hook, which receives pydhcp DHCPOptions and returns them, or any mapping of option names.

Source code in src/netboot/dhcp.py

def build_options(ctx: PixieContext, server: dhcpctl.DHCPServer) -> dhcpctl.DHCPOptions:
    """The options for this target on this server, in merge order.

    Later wins: zone defaults, the image's `dhcp_options`, the target's
    `dhcp_options`, then the `BuildDhcpOptions` hook, which receives pydhcp
    `DHCPOptions` and returns them, or any mapping of option names.
    """
    from . import PixieEvent

    layers = [lease.options for lease in _declared(ctx)]
    options = zone_defaults(ctx.dhcpzone, _codes(ctx)).copy(*layers)

    netboot = getattr(ctx, "_netboot_", None)
    if netboot is not None:
        options = _returned(
            "a BuildDhcpOptions hook",
            netboot.hook(
                PixieEvent.BuildDhcpOptions, options, target=ctx.target, server=server
            ),
            _codes(ctx),
        )
    return options

build_lease

build_lease(
    ctx: PixieContext, server: DHCPServer
) -> dhcpctl.DHCPLease

What this target is served on this server: its address, build_options, and the header fields the image or the target sets, the target's first.

Source code in src/netboot/dhcp.py

def build_lease(ctx: PixieContext, server: dhcpctl.DHCPServer) -> dhcpctl.DHCPLease:
    """What this target is served on this server: its address, `build_options`,
    and the header fields the image or the target sets, the target's first."""

    fields = {}
    for lease in _declared(ctx):
        if not lease.siaddr.is_unspecified:
            fields["siaddr"] = lease.siaddr
        for name in ("sname", "file"):
            if getattr(lease, name):
                fields[name] = getattr(lease, name)
    try:
        return dhcpctl.DHCPLease(
            ctx.target.ip or dhcpctl.UNSPECIFIED,
            options=build_options(ctx, server),
            strict=False,
            **fields,
        )
    except ValueError as exc:
        raise PixieConfigError(f"target {ctx.target._id!r}: {exc}") from exc

build_conditions

build_conditions(
    ctx: PixieContext,
) -> "dict[str, dhcpctl.Condition]"

Every dhcp_when condition that applies to this target, layered by name.

Zone, then image, then target -- later wins, by name, so a target redeclaring ipxe replaces the image's rather than adding a second one.

Source code in src/netboot/dhcp.py

def build_conditions(ctx: PixieContext) -> "dict[str, dhcpctl.Condition]":
    """Every `dhcp_when` condition that applies to this target, layered by name.

    Zone, then image, then target -- later wins, **by name**, so a target
    redeclaring `ipxe` replaces the image's rather than adding a second one.
    """
    conditions: "dict[str, dhcpctl.Condition]" = {}
    for source in (ctx.dhcpzone, ctx.image, ctx.target):
        if source is None:
            continue
        conditions.update(
            _config(
                dhcpctl.parse_conditions,
                getattr(source, "dhcp_when", None),
                getattr(source, "_id", source),
                _codes(ctx),
            )
        )
    return conditions

build_completion

build_completion(ctx: PixieContext) -> dhcpctl.Completion

How this target completes: zone, then image, then target -- later wins.

Source code in src/netboot/dhcp.py

def build_completion(ctx: PixieContext) -> dhcpctl.Completion:
    """How this target completes: zone, then image, then target -- later wins."""
    completion = dhcpctl.Completion()
    for source in (ctx.dhcpzone, ctx.image, ctx.target):
        if source is None:
            continue
        declared = _config(
            dhcpctl.parse_completion,
            getattr(source, "dhcp_complete", None),
            getattr(source, "_id", source),
            _codes(ctx),
        )
        if declared is not None:
            completion = declared
    return completion

condition_recipe

condition_recipe(
    ctx: "PixieContext",
    server: DHCPServer,
    condition: Condition,
) -> str

What server needs set up once for condition, in its own syntax.

Source code in src/netboot/dhcp.py

def condition_recipe(
    ctx: "PixieContext", server: dhcpctl.DHCPServer, condition: dhcpctl.Condition
) -> str:
    """What `server` needs set up once for `condition`, in its own syntax."""
    return server.condition_recipe(reservation_for(ctx, server), condition)

reservation_for

reservation_for(
    ctx: "PixieContext", server: DHCPServer
) -> dhcpctl.Reservation

This target's reservation on server, with ctx as its context.

Source code in src/netboot/dhcp.py

def reservation_for(
    ctx: "PixieContext", server: dhcpctl.DHCPServer
) -> dhcpctl.Reservation:
    """This target's reservation on `server`, with `ctx` as its context."""
    target = ctx.target
    mac = target.mac
    if mac is not None and str(mac) == dhcpctl.NULL_MAC:
        mac = None
    return dhcpctl.Reservation(
        target=dhcpctl.Target(
            name=str(target._id),
            mac=mac,
            hostname=str(getattr(target, "hostname", "") or ""),
        ),
        scope=scope_for(ctx.dhcpzone),
        lease=build_lease(ctx, server),
        conditions=build_conditions(ctx),
        completion=build_completion(ctx),
        context=ctx,
    )

arm

arm(ctx: 'PixieContext', server: DHCPServer) -> None

Arm server for this target.

Source code in src/netboot/dhcp.py

def arm(ctx: "PixieContext", server: dhcpctl.DHCPServer) -> None:
    """Arm `server` for this target."""
    with _lookup_errors():
        server.add(reservation_for(ctx, server))

disarm

disarm(ctx: 'PixieContext', server: DHCPServer) -> None

Disarm server for this target, honouring dhcp_complete:.

Source code in src/netboot/dhcp.py

def disarm(ctx: "PixieContext", server: dhcpctl.DHCPServer) -> None:
    """Disarm `server` for this target, honouring `dhcp_complete:`."""
    with _lookup_errors():
        server.complete(reservation_for(ctx, server))

Content repositories

netboot.content.Repository

Bases: Namespace

Where artifacts are served from.

services maps a scheme name to a base path or full URI; local is the filesystem view. address fills in the authority for a relative service path.

Source code in src/netboot/content/__init__.py

class Repository(_NS):
    """Where artifacts are served from.

    `services` maps a scheme name to a base path or full URI; `local` is the
    filesystem view. `address` fills in the authority for a relative service
    path.
    """

    address: Host
    services: dict[str, URIPath]
    local: "_ty.Optional[URIPath]"

    def __init__(self, **kwargs: object) -> None:
        self.services = {}
        kwargs.setdefault("address", None)
        kwargs.setdefault("local", None)
        super().__init__(**kwargs)
        if not isinstance(self.address, Host):
            self.address = Host(self.address)
        if self.local and not isinstance(self.local, URIPath):
            text = str(self.local)
            if len(text) > 1 and text[1] == ":" and text[0].isalpha():
                # A drive letter is not a URI scheme: "C:\\tftp" parsed as
                # scheme "c" produced a path nothing could read. Spell it as
                # the file URI it is.
                self.local = URIPath("file:///" + text.replace("\\", "/"))
            else:
                self.local = URIPath(text)

    def __getitem__(self, key: tuple[str, str]) -> URIPath:
        rel_path, service = key
        return self.get(rel_path, service=service)

    def __truediv__(self, key) -> "Repository":
        return self.joinpath(key)

    def joinpath(self, subpath: "str | None" = None) -> "Repository":
        """A copy of this repository with `subpath` appended to every service.

        `subpath` is always relative: `repo / "/abs"` extends the base like
        `repo / "abs"` does, rather than replacing it -- a leading slash in a
        config value is a typo, not a request to discard the repository root.
        """
        sub = str(subpath).strip("/") if subpath else None
        repo = Repository(address=self.address, services={}, local=self.local)
        for srvc, base in self.services.items():
            # Through URIPath, so a query string stays last.
            repo.services[srvc] = URIPath(str(base)) / sub if sub else base
        if self.local and sub:
            repo.local = self.local / sub
        return repo

    def get(self, *path: Union[RepoPath, str], service: str = None) -> URIPath:
        """Join `path` onto a service's base URI, or `None` if undefined."""
        path: RepoPath = RepoPath(
            *[(p if isinstance(p, Pathname) else RepoPath(p)).as_posix() for p in path]
        )
        rel_path = path.as_posix().lstrip("/")
        baseuri = self.service(service)
        if not baseuri:
            return None
        return baseuri / rel_path

    def _service_host(self, scheme: str) -> "tuple[str, _ty.Optional[int]]":
        """The authority for a relative service path: host and optional port.

        The port is split off **before** anything is resolved: an address like
        `mirror.example:8080` is a host and a port, and asking DNS for the whole
        string is a lookup that can only fail (slowly, and differently on every
        machine).

        TLS is the exception to resolving at all: a certificate is issued for a
        name, and a name-based virtual host needs one too, so `https` keeps the
        address as configured. Other schemes use the resolved IP, because a PXE
        client often has no working DNS when it fetches its boot files.

        The split is netimps' `split_host`, which netboot duplicated for a
        while. It returns a v6 literal **unbracketed**, which is what
        uripath's `Source` wants -- it brackets one itself.
        """
        address = str(self.address)
        if not address:
            LOGGER.warning(
                "repository has no address, so %s:// URLs are built without a "
                "host; give the repo an `address`, or write the service as a "
                "full URI",
                scheme,
            )
            return "", None
        try:
            host, port = split_host(address)
        except ValueError as exc:
            # `mirror.example:http` is a config mistake. The duplicate this
            # replaced treated it as a hostname, so it reached DNS as one.
            raise PixieConfigError(
                f"repository address {address!r} is not a host or host:port: {exc}"
            ) from exc
        if scheme == "https":
            return host, port
        # An unresolvable name still has to build a URL, so it is used as written.
        # netimps' shared cache answers the next URL for this name, miss included.
        ip = Host(host).ip(cache=True)
        return (str(ip) if ip else host), port

    def service(self, name: "str | None") -> "URIPath | None":
        """The base URI for a service name, `None` for `.local` when unset.

        A relative service path gets its authority from `.address`: resolved to an
        IP for every scheme except `https`, which keeps the configured name.
        """
        local = name is None
        if local:
            baseuri = self.local
            name = "file"
        else:
            baseuri = self.services.get(name, None)
        if baseuri is None:
            return None
        if not isinstance(baseuri, URIPath):
            baseuri = URIPath(baseuri)
        if not baseuri.source:
            # Only a relative service needs the address; a full URI has its own.
            host, port = ("", None) if local else self._service_host(name)
            # host=None writes `file:/srv/tftp`; "" would write an empty
            # authority (`file:///srv/tftp`) and change every URL built here.
            baseuri = baseuri.with_source(
                Source(scheme=name, userinfo=None, host=host or None, port=port)
            )

        return baseuri

address instance-attribute

address: Host

local instance-attribute

local: Optional[URIPath]

services instance-attribute

services: dict[str, URIPath] = {}

joinpath

joinpath(subpath: str | None = None) -> Repository

A copy of this repository with subpath appended to every service.

subpath is always relative: repo / "/abs" extends the base like repo / "abs" does, rather than replacing it -- a leading slash in a config value is a typo, not a request to discard the repository root.

Source code in src/netboot/content/__init__.py

def joinpath(self, subpath: "str | None" = None) -> "Repository":
    """A copy of this repository with `subpath` appended to every service.

    `subpath` is always relative: `repo / "/abs"` extends the base like
    `repo / "abs"` does, rather than replacing it -- a leading slash in a
    config value is a typo, not a request to discard the repository root.
    """
    sub = str(subpath).strip("/") if subpath else None
    repo = Repository(address=self.address, services={}, local=self.local)
    for srvc, base in self.services.items():
        # Through URIPath, so a query string stays last.
        repo.services[srvc] = URIPath(str(base)) / sub if sub else base
    if self.local and sub:
        repo.local = self.local / sub
    return repo

get

get(
    *path: Union[PosixPathname, str], service: str = None
) -> URIPath

Join path onto a service's base URI, or None if undefined.

Source code in src/netboot/content/__init__.py

def get(self, *path: Union[RepoPath, str], service: str = None) -> URIPath:
    """Join `path` onto a service's base URI, or `None` if undefined."""
    path: RepoPath = RepoPath(
        *[(p if isinstance(p, Pathname) else RepoPath(p)).as_posix() for p in path]
    )
    rel_path = path.as_posix().lstrip("/")
    baseuri = self.service(service)
    if not baseuri:
        return None
    return baseuri / rel_path

service

service(name: str | None) -> URIPath | None

The base URI for a service name, None for .local when unset.

A relative service path gets its authority from .address: resolved to an IP for every scheme except https, which keeps the configured name.

Source code in src/netboot/content/__init__.py

def service(self, name: "str | None") -> "URIPath | None":
    """The base URI for a service name, `None` for `.local` when unset.

    A relative service path gets its authority from `.address`: resolved to an
    IP for every scheme except `https`, which keeps the configured name.
    """
    local = name is None
    if local:
        baseuri = self.local
        name = "file"
    else:
        baseuri = self.services.get(name, None)
    if baseuri is None:
        return None
    if not isinstance(baseuri, URIPath):
        baseuri = URIPath(baseuri)
    if not baseuri.source:
        # Only a relative service needs the address; a full URI has its own.
        host, port = ("", None) if local else self._service_host(name)
        # host=None writes `file:/srv/tftp`; "" would write an empty
        # authority (`file:///srv/tftp`) and change every URL built here.
        baseuri = baseuri.with_source(
            Source(scheme=name, userinfo=None, host=host or None, port=port)
        )

    return baseuri

netboot.content.Resource

Bases: Namespace

A file in a repository: path within it, src naming the repo.

Source code in src/netboot/content/__init__.py

class Resource(_NS):
    """A file in a repository: `path` within it, `src` naming the repo."""

    path: Pathname
    src: str

    def __truediv__(self, key):
        return Resource(path=self.path / key, src=self.src)

    @classmethod
    def _parse_path(cls, path: Union[str, Pathname]):
        if not isinstance(path, Pathname):
            path = RepoPath(path)
        return path

path instance-attribute

path: Pathname

src instance-attribute

src: str

Templates

Rendering is tplx's; its API reference covers the loader, the engines and the engine contract.

Utilities

netboot.utils.misc

T module-attribute

T = TypeVar('T')

import_

import_(name: str) -> Any

Import "pkg.mod.attr" and return attr (splits on the last dot).

Source code in src/netboot/utils/misc.py

def import_(name: str) -> Any:
    """Import `"pkg.mod.attr"` and return `attr` (splits on the last dot)."""
    module, obj = name.rsplit(".", maxsplit=1)
    return getattr(import_module(module), obj)

parse_path

parse_path(path: str | Path) -> Path

Parse a configured path: a bare path is local, scheme: is a URI.

A single-letter scheme is a Windows drive, not a URI: C:\srv\tftp and C:/srv/tftp are local paths. Real URI schemes are at least two characters, so this costs nothing and stops every absolute Windows path from being parsed as a URI (and then failing with NotImplementedError when something tries to read it).

Source code in src/netboot/utils/misc.py

def parse_path(path: "str | Path") -> Path:
    """Parse a configured path: a bare path is local, ``scheme:`` is a URI.

    A single-letter scheme is a Windows drive, not a URI: ``C:\\srv\\tftp`` and
    ``C:/srv/tftp`` are local paths. Real URI schemes are at least two
    characters, so this costs nothing and stops every absolute Windows path
    from being parsed as a URI (and then failing with `NotImplementedError`
    when something tries to read it).
    """
    if isinstance(path, Path):
        return path
    text = str(path)
    return URIPath(text) if _URI_SCHEME.match(text) else LocalPath(text)

shell_quote

shell_quote(value: Any) -> Union[str, List[str]]

shlex.quote for a template: a POSIX shell reads the result as literal text.

A list or tuple is quoted element-wise and comes back as a list; anything else gives a str. A value that is not a string is stringified first, and None is the empty string (''). A value with nothing special to the shell comes back as it is.

Source code in src/netboot/utils/misc.py

def shell_quote(value: Any) -> "Union[str, List[str]]":
    """`shlex.quote` for a template: a POSIX shell reads the result as literal text.

    A list or tuple is quoted element-wise and comes back as a list; anything
    else gives a `str`. A value that is not a string is stringified first, and
    ``None`` is the empty string (``''``). A value with nothing special to the
    shell comes back as it is.
    """
    if isinstance(value, (list, tuple)):
        return [shell_quote(item) for item in value]
    return shlex.quote("" if value is None else str(value))

CLI entry point

netboot.cli

The pixie command line, built on duho.

main is the entry point of the pixie console script and of python -m netboot. Each built-in command is a class in a module of its own; a --cmdspath plugin may still be a module with run(netboot, args, conf).

Command

Bases: Cmd

A pixie command. The global options arrive from the root parser.

Source code in src/netboot/cli/_common.py

class Command(Cmd):
    """A `pixie` command. The global options arrive from the root parser."""

    #: The engine, built from the config on first use; a test may set it.
    _netboot_ = None

    @property
    def netboot(self) -> "Pixie":
        """The engine, from `--load-module`, `--config` and `--baseconfig`."""
        if self._netboot_ is None:
            from ._root import build_engine

            self._netboot_ = build_engine(self)[0]
        return self._netboot_

    def find(self, query: str) -> "PixieTarget | None":
        """The target `query` names, or None after logging why there is none.

        An ambiguous query is refused rather than guessed: arming the wrong
        machine is worse than arming none.
        """
        try:
            target = self.netboot.lookup_target(query)
        except LookupError as exc:
            LOGGER.error("%s", exc)
            return None
        if target is None:
            LOGGER.error("Target not found: %s", query)
        return target

netboot property

netboot: 'Pixie'

The engine, from --load-module, --config and --baseconfig.

find

find(query: str) -> 'PixieTarget | None'

The target query names, or None after logging why there is none.

An ambiguous query is refused rather than guessed: arming the wrong machine is worse than arming none.

Source code in src/netboot/cli/_common.py

def find(self, query: str) -> "PixieTarget | None":
    """The target `query` names, or None after logging why there is none.

    An ambiguous query is refused rather than guessed: arming the wrong
    machine is worse than arming none.
    """
    try:
        target = self.netboot.lookup_target(query)
    except LookupError as exc:
        LOGGER.error("%s", exc)
        return None
    if target is None:
        LOGGER.error("Target not found: %s", query)
    return target

main

main(
    argv: "_ty.Sequence[str] | None" = None,
    *,
    name: "str | None" = None
) -> int

Build the app, parse argv, and run the selected command.

name (default "pixie") is both the prog name and the environment prefix: settings are read through :class:duho.env.Env, so PIXIE_<KEY> variables apply. The resolved Env is attached to the dispatched instance as _env_.

Goes through :func:duho.run with exit="parse": it returns the status, except that --help, --version and a usage error raise SystemExit. A configuration or lookup error is reported as a single line and exits 2; the traceback is logged at DEBUG, so -v still shows it. Other exceptions propagate untouched.

Source code in src/netboot/cli/__init__.py

def main(
    argv: "_ty.Sequence[str] | None" = None,
    *,
    name: "str | None" = None,
) -> int:
    """Build the app, parse ``argv``, and run the selected command.

    ``name`` (default ``"pixie"``) is both the prog name and the environment
    prefix: settings are read through :class:`duho.env.Env`, so ``PIXIE_<KEY>``
    variables apply. The resolved ``Env`` is attached to the dispatched instance
    as ``_env_``.

    Goes through :func:`duho.run` with ``exit="parse"``: it returns the status,
    except that ``--help``, ``--version`` and a usage error raise ``SystemExit``.
    A configuration or lookup error is reported as a single line and exits 2;
    the traceback is logged at DEBUG, so ``-v`` still shows it. Other
    exceptions propagate untouched.
    """
    name = name or "pixie"
    # No companion `pixie_env` module: duho would import it from sys.path,
    # which under `python -m netboot` is the working directory.
    env = Env(name, autoload=False)
    # The CLI is an application, so it may quiet noisy dependencies; importing
    # the library does not do this for its host.
    quiet_noisy_dependencies()
    try:
        commands = _discover(argv)
    except _USER_ERRORS as exc:
        # Discovery happens before duho dispatches, so duho does not see this.
        _report_error(str(exc), 2, exc)
        return 2
    return _run(
        Pixie_,
        argv,
        exit="parse",
        commands=commands,
        name=name,
        description=Pixie_.__doc__,
        env=env,
        dispatch=_dispatch,
    )

Errors

netboot.PixieError

Bases: Exception

Base for the errors netboot raises on purpose.

Separates "the operator gave us something unusable" from "netboot has a bug": the CLI reports the former as a one-line message and keeps the traceback for the latter.

Source code in src/netboot/exceptions.py

class PixieError(Exception):
    """Base for the errors netboot raises on purpose.

    Separates "the operator gave us something unusable" from "netboot has a
    bug": the CLI reports the former as a one-line message and keeps the
    traceback for the latter.
    """

netboot.PixieLookupError

Bases: PixieError, LookupError

A target, image or zone could not be resolved, or the query was ambiguous.

Source code in src/netboot/exceptions.py

class PixieLookupError(PixieError, LookupError):
    """A target, image or zone could not be resolved, or the query was ambiguous."""

netboot.PixieConfigError

Bases: PixieError, ValueError

The configuration cannot be used as given.

Source code in src/netboot/exceptions.py

class PixieConfigError(PixieError, ValueError):
    """The configuration cannot be used as given."""