nebius.aio.token.federation_account.FederationBearer

class documentation

class FederationBearer(ParentBearer): (source)

Constructor: FederationBearer(profile_name, client_id, federation_endpoint, federation_id, ...)

Bearer composing interactive federation auth with a renewable cache.

The bearer constructs an internal nebius.aio.token.federation_bearer.Bearer configured with the provided parameters and wraps it in an AsynchronousRenewableFileCacheBearer to provide persistent caching and background refresh.

Behavior and authorization flow

When a token is requested via nebius.aio.token.token.Receiver.fetch the following sequence may occur:

The cache is consulted for a fresh token. If a cached token is valid it

is returned immediately.
If renewal is required the underlying interactive flow is invoked. This
flow is implemented by nebius.aio.token.federation_bearer.Receiver and will typically:
- construct an authorization URL and attempt to open the user's
  
  browser (unless no_browser_open is set), or return the URL so callers can display it.
- block while waiting for the user to complete the authorization in the
  
  browser (for example by granting access). This means the fetch may hang until the user completes the flow.
- receive an access token and expiration and return it to the caller.

Where the URL is shown

The interactive helper logs the authorization URL so it appears in the

application's logs.
If a writer stream is provided the helper will also write the URL and

short instructions to that stream (e.g. stdout), which is useful for headless environments.

Example

Construct a bearer and use it to initialize the SDK:

from nebius.sdk import SDK
from nebius.aio.token.federation_account import FederationBearer
import sys

sdk = SDK(credentials=FederationBearer(
    profile_name="not-a-cli-profile",
    client_id="my-client-id",
    federation_endpoint="auth.eu.nebius.com",
    federation_id="federation-e00my-federation",
    writer=sys.stdout,
    no_browser_open=True,
))

Parameters
profile_name	Human-readable profile name included in the bearer's `nebius.aio.token.token.Bearer.name` and used as part of the cache key.
client_id	OAuth2 client identifier used by the interactive flow.
federation_endpoint	Federation endpoint URL.
federation_id	Identifier of the federation configuration.
writer	Optional text stream to which the interactive helper will write the authorization URL and short instructions.
no_browser_open	When true the helper will not attempt to open a browser automatically and will instead rely on the provided `writer` or logs to surface the URL to the user.
timeout	Timeout forwarded to synchronous refresh requests.
max_retries	Default maximum retry attempts for the receiver.
initial_safety_margin	Safety margin used when deciding whether a cached token is considered too close to expiry. May be a timedelta or seconds value.
retry_safety_margin	Safety margin subtracted when computing the initial retry interval for the background loop.
lifetime_safe_fraction	Fraction of the token lifetime used to schedule proactive refreshes.
initial_retry_timeout	Initial backoff timeout used after failures.
max_retry_timeout	Maximum backoff timeout.
retry_timeout_exponent	Exponent used for exponential backoff.
file_cache_throttle	Throttle interval passed to the file cache layer to reduce disk reads.
ssl_ctx	Optional SSL context used for HTTPS requests.

Method	`__init__`	Initialize the federation bearer with renewable file cache.
Method	`receiver`	Return a receiver from the underlying renewable file cache bearer.
Property	`wrapped`	Return the wrapped bearer instance.
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, profile_name: str, client_id: str, federation_endpoint: str, federation_id: str, writer: TextIO | None = None, no_browser_open: bool = False, timeout: timedelta = timedelta(minutes=5), max_retries: int = 2, initial_safety_margin: timedelta | float | None = timedelta(hours=2), retry_safety_margin: timedelta = timedelta(hours=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, file_cache_throttle: timedelta | float = timedelta(minutes=5), ssl_ctx: SSLContext | None = None): (source) ¶

Initialize the federation bearer with renewable file cache.

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

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

Return a receiver from the underlying renewable file cache bearer.

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

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

Return the wrapped bearer instance.

_source = (source) ¶

Undocumented