flayer.extensions

Module source

flayer.extensions

Additive typed extension contracts, static metadata discovery, and explicit loading.

PROFILE_GROUP module-attribute

PROFILE_GROUP = 'flayer.profiles.v1'

PROVIDER_GROUP module-attribute

PROVIDER_GROUP = 'flayer.providers.v1'

CompiledProfile

Bases: Protocol

Validated profile adapter that preserves existing artifact and lifecycle boundaries.

Identity property

Identity: StackIdentity

Return the exact identity validated during profile compilation.

BuildServerBundle

BuildServerBundle() -> ArtifactBundle

Generate owned server initialization bytes without writing or deploying them.

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

def BuildServerBundle(self) -> ArtifactBundle:
    """Generate owned server initialization bytes without writing or deploying them."""

    ...

RenderPlan

RenderPlan(*, user_data_file: str | Path) -> str

Render lifecycle TOML after verifying the exact owned server artifact.

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

def RenderPlan(self, *, user_data_file: str | Path) -> str:
    """Render lifecycle TOML after verifying the exact owned server artifact."""

    ...

ExtensionDescriptor dataclass

ExtensionDescriptor(kind: ExtensionKind, plugin_id: str, distribution: str, version: str, group: str, target: str, metadata_directory: Path | None = None, metadata_sha256: str | None = None)

A static identity; external metadata snapshots are not package authenticity proof.

kind instance-attribute

kind: ExtensionKind

plugin_id instance-attribute

plugin_id: str

distribution instance-attribute

distribution: str

version instance-attribute

version: str

group instance-attribute

group: str

target instance-attribute

target: str

metadata_directory class-attribute instance-attribute

metadata_directory: Path | None = field(default=None, repr=False)

metadata_sha256 class-attribute instance-attribute

metadata_sha256: str | None = field(default=None, repr=False)

ExtensionError

Bases: ContractError

An extension violates discovery, identity, compatibility, or loading policy.

ExtensionKind

Bases: StrEnum

The two implemented extension contracts; their ABI is encoded in the group.

PROVIDER class-attribute instance-attribute

PROVIDER = 'provider'

PROFILE class-attribute instance-attribute

PROFILE = 'profile'

LifecycleProviderExtension

Bases: ProviderExtension, Protocol

Optional separate factory for the existing owned resource lifecycle contract.

CreateLifecycleProvider

CreateLifecycleProvider(context: ProviderContext) -> LifecycleProvider

Construct a mutation-capable adapter without performing provider operations.

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

def CreateLifecycleProvider(self, context: ProviderContext) -> LifecycleProvider:
    """Construct a mutation-capable adapter without performing provider operations."""

    ...

ProfileContext dataclass

ProfileContext(identity: StackIdentity)

Exact stack ownership context; profile configuration and artifacts stay separate.

identity instance-attribute

identity: StackIdentity

ProfileExtension

Bases: Protocol

Compile explicit profile configuration using its own strict versioned schema.

Descriptor property

Descriptor: ExtensionDescriptor

Return the exact static identity registered before factory loading.

CompileProfile

CompileProfile(configuration: Mapping[str, object], context: ProfileContext) -> CompiledProfile

Validate configuration and exact ownership without accessing infrastructure.

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

def CompileProfile(
    self, configuration: Mapping[str, object], context: ProfileContext,
) -> CompiledProfile:
    """Validate configuration and exact ownership without accessing infrastructure."""

    ...

ProviderContext dataclass

ProviderContext(scope_id: str, authentication_reference: str, command_timeout: float = 45.0, inventory_limit: int = 10000)

Explicit scope, external CLI authentication reference, and bounded execution limits.

scope_id instance-attribute

scope_id: str

authentication_reference instance-attribute

authentication_reference: str

command_timeout class-attribute instance-attribute

command_timeout: float = 45.0

inventory_limit class-attribute instance-attribute

inventory_limit: int = 10000

ProviderExtension

Bases: Protocol

Factory for the existing read-only provider contract, without mandatory mutation.

Descriptor property

Descriptor: ExtensionDescriptor

Return the exact static identity registered before factory loading.

CreateProvider

CreateProvider(context: ProviderContext) -> CloudProvider

Construct a scoped read-only adapter without probing or mutating infrastructure.

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

def CreateProvider(self, context: ProviderContext) -> CloudProvider:
    """Construct a scoped read-only adapter without probing or mutating infrastructure."""

    ...

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

BuiltinDescriptors

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

Return static descriptors without importing provider or profile implementations.

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

def BuiltinDescriptors() -> tuple[ExtensionDescriptor, ...]:
    """Return static descriptors without importing provider or profile implementations."""

    return (YANDEX_READ_ONLY, YANDEX_LIFECYCLE, SECURE_GATEWAY)

DiscoverExtensions

DiscoverExtensions(metadata_directories: Sequence[Path]) -> tuple[ExtensionDescriptor, ...]

Parse selected metadata without importing code, calling finders, or searching paths.

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

def DiscoverExtensions(metadata_directories: Sequence[Path]) -> tuple[ExtensionDescriptor, ...]:
    """Parse selected metadata without importing code, calling finders, or searching paths."""

    if (
        not isinstance(metadata_directories, (tuple, list))
        or len(metadata_directories) > MAX_METADATA_DIRECTORIES
        or any(not isinstance(directory, Path) for directory in metadata_directories)
        or len(set(metadata_directories)) != len(metadata_directories)
    ):
        raise ExtensionError("Discovery requires bounded unique explicit metadata directories")

    descriptors: list[ExtensionDescriptor] = []
    identities: set[tuple[ExtensionKind, str]] = set()

    for directory in metadata_directories:
        contents, digest = ReadMetadataSnapshot(directory)

        try:
            metadata = BytesParser(policy=policy.default).parsebytes(contents[0])
            names = metadata.get_all("Name", [])
            versions = metadata.get_all("Version", [])

            if len(names) != 1 or len(versions) != 1 or metadata.defects:
                raise ExtensionError("Distribution metadata requires one valid Name and Version")

            distribution = str(names[0]).lower().replace("_", "-").replace(".", "-")
            version = str(versions[0])
            parser = _EntryPointsParser(interpolation=None, strict=True, delimiters=("=",))
            parser.read_string(contents[1].decode("utf-8"))

            if parser.defaults():
                raise ExtensionError("Entry-point metadata cannot contain default entries")

            for group in parser.sections():
                if not group.startswith("flayer."):
                    continue

                if group not in {PROVIDER_GROUP, PROFILE_GROUP}:
                    raise ExtensionError("Entry-point metadata uses an unsupported F-Layer group")

                kind = ExtensionKind.PROVIDER if group == PROVIDER_GROUP else ExtensionKind.PROFILE

                for plugin_id, target in parser.items(group):
                    descriptor = ExtensionDescriptor(
                        kind, plugin_id, distribution, version, group, target, directory, digest,
                    )
                    identity = (kind, plugin_id)

                    if identity in identities or len(descriptors) >= MAX_EXTENSIONS:
                        raise ExtensionError("Extension metadata contains duplicate or excess identities")

                    identities.add(identity)
                    descriptors.append(descriptor)

        except (UnicodeError, configparser.Error, ValueError):
            raise ExtensionError("Extension metadata is malformed or incompatible") from None

    return tuple(sorted(descriptors, key=lambda item: (item.kind.value, item.plugin_id)))