nebius.aio.token.impersonated.Receiver

class documentation

class Receiver(ParentReceiver): (source)

Constructor: Receiver(service_account_id, source, service, max_retries, ...)

View In Hierarchy

Receiver that exchanges an actor token for an impersonated token.

Method __init__ Create a receiver bound to the source token receiver.
Method can_retry Allow the request layer to retry a failed impersonation fetch.
Async Method _exchange Exchange actor_token for the configured service account token.
Async Method _fetch Fetch the source token and exchange it for an impersonated token.
Method _should_retry_actor Undocumented
Instance Variable _max_retries Undocumented
Instance Variable _metrics Undocumented
Instance Variable _service_account_id Undocumented
Instance Variable _source Undocumented
Instance Variable _svc Undocumented
Instance Variable _trial Undocumented

Inherited from Receiver:

Async Method fetch Fetch a token and record it as the latest value.
Property latest Return the latest fetched token or None.
Instance Variable _latest The most recently fetched token, or None if no token has been fetched yet. Implementations may update this to support inspection or caching by callers.
def __init__(self, service_account_id: str, source: ParentReceiver, service: TokenExchangeServiceClient | Awaitable[TokenExchangeServiceClient], max_retries: int = 2, metrics: AuthMetricsLike = None): (source)

Create a receiver bound to the source token receiver.

def can_retry(self, err: Exception, options: dict[str, str] | None = None) -> bool: (source)
overrides nebius.aio.token.token.Receiver.can_retry

Allow the request layer to retry a failed impersonation fetch.

async def _exchange(self, actor_token: str, now: datetime, timeout: float | None) -> Token: (source)

Exchange actor_token for the configured service account token.

async def _fetch(self, timeout: float | None = None, options: dict[str, str] | None = None) -> Token: (source)
overrides nebius.aio.token.token.Receiver._fetch

Fetch the source token and exchange it for an impersonated token.

def _should_retry_actor(self, err: RequestError, options: dict[str, str] | None) -> bool: (source)

Undocumented

_max_retries = (source)

Undocumented

_metrics = (source)

Undocumented

_service_account_id = (source)

Undocumented

_source = (source)

Undocumented

_svc = (source)

Undocumented

_trial: int = (source)

Undocumented