flayer.providers.yandex

Module source

flayer.providers.yandex

Read-only Yandex Cloud CLI adapter with bounded commands and sanitized failures.

PROVIDER_ID module-attribute

PROVIDER_ID = 'yandex-cloud'

RESOURCE_COMMANDS module-attribute

RESOURCE_COMMANDS: dict[ResourceKind, tuple[str, str]] = {ResourceKind.INSTANCE: ('compute', 'instance'), ResourceKind.DISK: ('compute', 'disk'), ResourceKind.NETWORK: ('vpc', 'network'), ResourceKind.SUBNET: ('vpc', 'subnet'), ResourceKind.ADDRESS: ('vpc', 'address'), ResourceKind.SECURITY_GROUP: ('vpc', 'security-group')}

ERROR_MARKERS module-attribute

ERROR_MARKERS: tuple[tuple[ProviderErrorCode, tuple[str, ...]], ...] = ((ProviderErrorCode.AUTHENTICATION, ('unauthenticated', 'invalid token', 'token expired')), (ProviderErrorCode.PERMISSION_DENIED, ('permissiondenied', 'permission_denied', 'forbidden')), (ProviderErrorCode.NOT_FOUND, ('notfound', 'not_found', 'not found')), (ProviderErrorCode.TIMEOUT, ('deadlineexceeded', 'deadline_exceeded', 'timed out')), (ProviderErrorCode.THROTTLED, ('resourceexhausted', 'resource_exhausted', 'too many requests')), (ProviderErrorCode.CONFLICT, ('alreadyexists', 'already_exists', 'conflict')))

YandexCloudSettings dataclass

YandexCloudSettings(folder_id: str, profile: str = 'default', executable: str = 'yc', command_timeout: float = 45.0, inventory_limit: int = 10000)

Explicit folder/profile configuration with no stored tokens or secret keys.

folder_id instance-attribute

folder_id: str

profile class-attribute instance-attribute

profile: str = 'default'

executable class-attribute instance-attribute

executable: str = 'yc'

command_timeout class-attribute instance-attribute

command_timeout: float = 45.0

inventory_limit class-attribute instance-attribute

inventory_limit: int = 10000

CommandResult dataclass

CommandResult(return_code: int, stdout: str = '', stderr: str = '')

An ephemeral command result whose representation omits potentially secret output.

return_code instance-attribute

return_code: int

stdout class-attribute instance-attribute

stdout: str = field(default='', repr=False)

stderr class-attribute instance-attribute

stderr: str = field(default='', repr=False)

CommandRunner

Bases: Protocol

Injectable execution boundary for deterministic tests and alternative transports.

Run

Run(command: tuple[str, ...], timeout: float) -> CommandResult

Execute one argument vector without modifying its order or scope.

Source code in installed/flayer/providers/yandex.py

def Run(self, command: tuple[str, ...], timeout: float) -> CommandResult:
    """Execute one argument vector without modifying its order or scope."""

    ...

SubprocessCommandRunner

Execute the configured CLI directly without a shell or interactive standard input.

Run

Run(command: tuple[str, ...], timeout: float) -> CommandResult

Capture output transiently while enforcing a per-command timeout.

Source code in installed/flayer/providers/yandex.py

def Run(self, command: tuple[str, ...], timeout: float) -> CommandResult:
    """Capture output transiently while enforcing a per-command timeout."""

    result = subprocess.run(
        command,
        check=False,
        stdin=subprocess.DEVNULL,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        shell=False,
        timeout=timeout,
    )

    return CommandResult(result.returncode, result.stdout, result.stderr)

YandexCloudProvider

YandexCloudProvider(settings: YandexCloudSettings, runner: CommandRunner | None = None)

Normalize CLI inventory inside one folder; expose no cloud mutation methods.

Construct a provider without executing commands or reading credential files.

Source code in installed/flayer/providers/yandex.py

def __init__(
    self, settings: YandexCloudSettings, runner: CommandRunner | None = None
) -> None:
    """Construct a provider without executing commands or reading credential files."""

    self._settings = settings
    self._runner = runner if runner is not None else SubprocessCommandRunner()
    self._identity = ProviderIdentity(PROVIDER_ID, settings.folder_id, "cli-profile")

Identity property

Identity: ProviderIdentity

Return the non-secret identity used to validate all resources and references.

Capabilities property

Capabilities: frozenset[ProviderCapability]

Declare implemented read operations without inferring lifecycle capability.

CheckAvailability

CheckAvailability() -> ProviderStatus

Check CLI execution locally; no token, profile or infrastructure read occurs.

Source code in installed/flayer/providers/yandex.py

def CheckAvailability(self) -> ProviderStatus:
    """Check CLI execution locally; no token, profile or infrastructure read occurs."""

    try:
        self._Run(("--version",), "availability", scoped=False)

    except ProviderError as error:
        return ProviderStatus(self.Identity, False, error_code=error.code)

    return ProviderStatus(self.Identity, True)

CheckAuthentication

CheckAuthentication() -> ProviderStatus

Verify access to the selected folder through the CLI's existing credentials.

Source code in installed/flayer/providers/yandex.py

def CheckAuthentication(self) -> ProviderStatus:
    """Verify access to the selected folder through the CLI's existing credentials."""

    try:
        payload = self._ReadJson(
            ("resource-manager", "folder", "get", "--id", self._settings.folder_id),
            "authentication",
        )

        if not isinstance(payload, dict) or payload.get("id") != self._settings.folder_id:
            raise ProviderError(ProviderErrorCode.SCOPE_MISMATCH, "authentication")

    except ProviderError as error:
        available = error.code != ProviderErrorCode.UNAVAILABLE

        return ProviderStatus(self.Identity, available, False, error.code)

    return ProviderStatus(self.Identity, True, True)

ListResources

ListResources(kind: ResourceKind) -> tuple[ProviderResource, ...]

Read one resource kind, rejecting malformed or possibly truncated inventory.

Source code in installed/flayer/providers/yandex.py

def ListResources(self, kind: ResourceKind) -> tuple[ProviderResource, ...]:
    """Read one resource kind, rejecting malformed or possibly truncated inventory."""

    command = self._CommandForKind(kind)
    payload = self._ReadJson(
        (*command, "list", "--limit", str(self._settings.inventory_limit)), "list"
    )

    if not isinstance(payload, list):
        raise ProviderError(ProviderErrorCode.INVALID_RESPONSE, "list")

    if len(payload) >= self._settings.inventory_limit:
        raise ProviderError(ProviderErrorCode.INCOMPLETE_INVENTORY, "list")

    resources = tuple(self._NormalizeResource(item, kind, "list") for item in payload)

    if len({resource.reference.resource_id for resource in resources}) != len(resources):
        raise ProviderError(ProviderErrorCode.INVALID_RESPONSE, "list")

    return tuple(sorted(resources, key=lambda item: item.reference.resource_id))

GetResource

GetResource(reference: ResourceReference) -> ProviderResource

Read by ID and revalidate returned scope because vendor ID lookup is global.

Source code in installed/flayer/providers/yandex.py

def GetResource(self, reference: ResourceReference) -> ProviderResource:
    """Read by ID and revalidate returned scope because vendor ID lookup is global."""

    if (
        reference.provider_id != self.Identity.provider_id
        or reference.scope_id != self.Identity.scope_id
    ):
        raise ProviderError(ProviderErrorCode.SCOPE_MISMATCH, "get")

    command = self._CommandForKind(reference.kind)
    payload = self._ReadJson((*command, "get", "--id", reference.resource_id), "get")
    resource = self._NormalizeResource(payload, reference.kind, "get")

    if resource.reference != reference:
        raise ProviderError(ProviderErrorCode.SCOPE_MISMATCH, "get")

    return resource

DiscoverInventory

DiscoverInventory() -> tuple[ProviderResource, ...]

Return a deterministic complete inventory, never a partial success snapshot.

Source code in installed/flayer/providers/yandex.py

def DiscoverInventory(self) -> tuple[ProviderResource, ...]:
    """Return a deterministic complete inventory, never a partial success snapshot."""

    resources: list[ProviderResource] = []

    for kind in ResourceKind:
        resources.extend(self.ListResources(kind))

    return tuple(resources)