nebius.aio.token.service_account.ServiceAccountBearer

class documentation

class ServiceAccountBearer(ParentBearer): (source)

Constructor: ServiceAccountBearer(service_account, channel, private_key, public_key_id, ...)

Bearer that obtains tokens using a service account.

The class composes an exchangeable bearer (that performs the token exchange), wraps it into a renewable bearer (to handle background token refresh) and finally assigns a stable name using NamedBearer.

The chain from the outermost to innermost is as follows:

NamedBearer ->

RenewableBearer ->

ExchangeableBearer ->

ServiceAccountReader ->

ServiceAccount

Example

Construct a bearer and use it to initialize the SDK:

from asyncio import Future
from nebius.sdk import SDK
from nebius.aio.token.service_account import ServiceAccountBearer
from cryptography.hazmat.primitives.serialization import load_pem_private_key

with open("/path/to/private_key.pem", "rb") as fh:
    private_key = load_pem_private_key(fh.read(), password=None)

# Create a future for the channel that will be resolved with the SDK
channel_future = Future()

sdk = SDK(credentials=ServiceAccountBearer(
    "service-account-id",
    private_key=private_key,
    public_key_id="public-key-id",
    channel=channel_future,
))

# Resolve the future with the newly created SDK
channel_future.set_result(sdk)

Parameters
service_account	Service account credentials used to obtain tokens. May be a `ServiceAccountReader`, a `ServiceAccount` or a string service account id.
channel	A channel used to perform the token exchange. This channel must be provided before any token fetch operation, or a `DeferredChannel` may be used to set the channel asynchronously. If neither is provided, token fetch operations will fail until `set_channel` is called.
private_key	When `service_account` is a string id, this private key is used to sign token exchange requests. Must not be provided if the service account is provided as a `ServiceAccount` or `ServiceAccountReader`.
public_key_id	When `service_account` is a string id, this is the public key ID corresponding to the private key. Must not be provided if the service account is provided as a `ServiceAccount` or `ServiceAccountReader`.
max_retries	Maximum number of retries for token fetch operations.
lifetime_safe_fraction	Fraction of token lifetime considered safe to use before triggering a refresh.
initial_retry_timeout	Initial delay between retry attempts for refresh operations.
max_retry_timeout	Maximum delay between retry attempts for refresh operations.
retry_timeout_exponent	Exponential backoff exponent for retry delays.
refresh_request_timeout	Timeout for individual token refresh requests.

Method	`__init__`	Initialize a service-account based bearer.
Method	`receiver`	Calls the receiver of the underlying bearer `NamedBearer`.
Method	`set_channel`	Attach a concrete channel to the exchangeable bearer.
Property	`wrapped`	Returns the outermost underlying bearer `NamedBearer`.
Instance Variable	`_exchangeable`	Undocumented
Instance Variable	`_source`	Undocumented

Inherited from Bearer:

Async Method	`close`	Close the bearer and any wrapped resources.
Property	`name`	Optional human-readable name for the bearer.

def __init__(self, service_account: ServiceAccountReader | ServiceAccount | str, channel: ClientChannelInterface | DeferredChannel | None = None, private_key: RSAPrivateKey | None = None, public_key_id: str | None = None, max_retries: int = 2, lifetime_safe_fraction: float = 0.9, initial_retry_timeout: timedelta = timedelta(seconds=1), max_retry_timeout: timedelta = timedelta(minutes=1), retry_timeout_exponent: float = 1.5, refresh_request_timeout: timedelta = timedelta(seconds=5)): (source) ¶

Initialize a service-account based bearer.

This is essentially a convenience wrapper that composes several internal bearer implementations to provide a ready-to-use bearer that fetches tokens using service account credentials and is being conveniently named with the service account parameters.

Important note: When constructing the bearer using a dynamic ServiceAccountReader, the name of the bearer will reflect the service account as read during construction time. If the reader returns different service accounts on subsequent reads, the name will not reflect those changes.

def receiver(self) -> Receiver: (source) ¶

overrides nebius.aio.token.token.Bearer.receiver

Calls the receiver of the underlying bearer NamedBearer.

def set_channel(self, channel: ClientChannelInterface): (source) ¶

Attach a concrete channel to the exchangeable bearer.

This function must be used when a channel was not available at construction and a DeferredChannel was not provided.

Parameters
channel:`ClientChannelInterface`	The concrete channel to attach.

@property
wrapped: ParentBearer | None = (source) ¶

overrides nebius.aio.token.token.Bearer.wrapped

Returns the outermost underlying bearer NamedBearer.

_exchangeable = (source) ¶

Undocumented

_source = (source) ¶

Undocumented