flayer.extensions.registry

Module source

flayer.extensions.registry

Explicit registry and opt-in loading of metadata-pinned, owned extension sources.

MAX_MODULE_BYTES module-attribute

MAX_MODULE_BYTES = 1048576

MAX_MODULE_DEPTH module-attribute

MAX_MODULE_DEPTH = 16

ExtensionRegistry

ExtensionRegistry(descriptors: Sequence[ExtensionDescriptor] = (), *, include_builtins: bool = True)

A bounded explicit catalog; neither construction nor lookup imports external code.

Validate complete registration before publishing any catalog entries.

Source code in installed/flayer/extensions/registry.py

def __init__(
    self, descriptors: Sequence[ExtensionDescriptor] = (), *, include_builtins: bool = True,
) -> None:
    """Validate complete registration before publishing any catalog entries."""

    if type(include_builtins) is not bool or not isinstance(descriptors, (tuple, list)):
        raise ExtensionError("Registry requires explicit bounded descriptor registration")

    entries = (*BuiltinDescriptors(), *descriptors) if include_builtins else tuple(descriptors)

    if len(entries) > MAX_EXTENSIONS:
        raise ExtensionError("Registry exceeds its extension bound")

    registered: dict[tuple[ExtensionKind, str], ExtensionDescriptor] = {}

    for descriptor in entries:
        if not isinstance(descriptor, ExtensionDescriptor):
            raise ExtensionError("Registry accepts only validated ExtensionDescriptor objects")

        identity = (descriptor.kind, descriptor.plugin_id)

        if identity in registered:
            raise ExtensionError("Duplicate extension identity would shadow a registered extension")

        if descriptor.metadata_directory is None and descriptor not in BuiltinDescriptors():
            raise ExtensionError("Registry rejects invented or incompatible built-in identities")

        registered[identity] = descriptor

    self._descriptors = registered
    self._loaded: dict[ExtensionDescriptor, ProviderExtension | ProfileExtension] = {}

Descriptors

Descriptors() -> tuple[ExtensionDescriptor, ...]

Return immutable descriptors in deterministic kind and identifier order.

Source code in installed/flayer/extensions/registry.py

def Descriptors(self) -> tuple[ExtensionDescriptor, ...]:
    """Return immutable descriptors in deterministic kind and identifier order."""

    return tuple(sorted(self._descriptors.values(), key=lambda item: (item.kind.value, item.plugin_id)))

Get

Get(kind: ExtensionKind, plugin_id: str) -> ExtensionDescriptor

Select one registered identity without imports or capability probes.

Source code in installed/flayer/extensions/registry.py

def Get(self, kind: ExtensionKind, plugin_id: str) -> ExtensionDescriptor:
    """Select one registered identity without imports or capability probes."""

    if not isinstance(kind, ExtensionKind) or not isinstance(plugin_id, str):
        raise ExtensionError("Registry lookup requires an exact extension kind and identifier")

    descriptor = self._descriptors.get((kind, plugin_id))

    if descriptor is None:
        raise ExtensionError("Extension identity is not registered")

    return descriptor

Load

Load(descriptor: ExtensionDescriptor, *, allowed: Sequence[ExtensionDescriptor] = ()) -> ProviderExtension | ProfileExtension

Load an exact registered factory; external code requires an exact descriptor pin.

Source code in installed/flayer/extensions/registry.py

def Load(
    self, descriptor: ExtensionDescriptor, *, allowed: Sequence[ExtensionDescriptor] = (),
) -> ProviderExtension | ProfileExtension:
    """Load an exact registered factory; external code requires an exact descriptor pin."""

    if (
        not isinstance(descriptor, ExtensionDescriptor)
        or self.Get(descriptor.kind, descriptor.plugin_id) != descriptor
        or not isinstance(allowed, (tuple, list)) or len(allowed) > MAX_EXTENSIONS
        or any(not isinstance(item, ExtensionDescriptor) for item in allowed)
        or len(set(allowed)) != len(allowed)
    ):
        raise ExtensionError("Loading requires an exact registered descriptor and bounded unique pins")

    if descriptor.metadata_directory is not None and descriptor not in allowed:
        raise ExtensionError("External loading requires an explicit exact descriptor allowlist")

    if descriptor in self._loaded:
        return self._loaded[descriptor]

    loaded = (
        _ValidateExtension(LoadBuiltin(descriptor), descriptor)
        if descriptor.metadata_directory is None else _LoadFactory(descriptor)
    )
    self._loaded[descriptor] = loaded

    return loaded