nebius.aio.token.federation

class documentation

class Receiver(ParentReceiver): (source)

Constructor: Receiver(client_id, federation_endpoint, federation_id, writer, ...)

Receiver that runs the federation interactive authorization.

Interactive authorization flow (what happens on `fetch`)

Receiver._fetch calls nebius.aio.token.federation_bearer.auth.authorize. This function performs the interactive federation handshake:

The helper constructs an authorization URL and either attempts to open

the user's default browser or returns the URL for the caller to display.
The helper will block while waiting for the user to complete the

authorization in the browser (for example by granting access). This means Receiver._fetch may hang until the user completes the flow.
The helper receives an authorization response containing an access token

and expiry. The receiver converts this into a nebius.aio.token.token.Token and returns it to the caller.

Where the URL is shown

The authorization URL is logged at INFO/DEBUG levels so it appears in the

project's logs.
If Receiver or Bearer has a writer, the helper writes the URL and short instructions to it. The caller can then show them to the user, for example on stdout.

Blocking considerations

Because the flow waits for the user's action, callers should be aware that fetch can block for an extended period. Use reasonable timeouts when calling Receiver._fetch or invoke the bearer within a background task if you need the application to remain responsive.

Parameters
client_id	OAuth2 client identifier used during authorization.
federation_endpoint	URL of the federation endpoint.
federation_id	Identifier of the users federation.
writer	Optional text writer used to display instructions to the user.
no_browser_open	If true the implementation will not attempt to open a browser automatically. It writes the URL to `writer` and the logs.
ssl_ctx	Optional SSL context to use for HTTPS requests.
metrics	Optional auth metrics callbacks used to record the interactive token acquisition result and token lifetime.

Method	`__init__`	Create a federation interactive receiver.
Method	`can_retry`	Decide whether an authorization error should be retried.
Async Method	`_fetch`	Execute the interactive authorization flow and return a Token.
Instance Variable	`_client_id`	Undocumented
Instance Variable	`_federation_endpoint`	Undocumented
Instance Variable	`_federation_id`	Undocumented
Instance Variable	`_metrics`	Undocumented
Instance Variable	`_no_browser_open`	Undocumented
Instance Variable	`_ssl_ctx`	Undocumented
Instance Variable	`_writer`	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, client_id: str, federation_endpoint: str, federation_id: str, writer: TextIO | None = None, no_browser_open: bool = False, ssl_ctx: SSLContext | None = None, metrics: AuthMetricsLike = None, provider: str | None = None): (source) ¶

Create a federation interactive receiver.

def can_retry(self, err: Exception, options: dict[str, str] | None = None) -> bool: (source) ¶

overrides nebius.aio.token.token.Receiver.can_retry

Decide whether an authorization error should be retried.

The current implementation always permits retries. This is a simple policy because interactive flows commonly allow a user to retry the operation (for example if they decline the prompt initially).

Parameters
err:`Exception`	The exception raised during fetch.
options:`dict[str, str] \| None`	Unused; present for API compatibility.
Returns
`bool`	Always `True` in the current implementation.

async def _fetch(self, timeout: float | None = None, options: dict[str, str] | None = None) -> Token: (source) ¶

overrides nebius.aio.token.token.Receiver._fetch

Execute the interactive authorization flow and return a Token.

The implementation imports and calls nebius.aio.token.federation_bearer.auth.authorize to perform the protocol. The returned access token and expiration information are converted into a nebius.aio.token.token.Token.

Parameters
timeout:`float \| None`	Optional timeout in seconds forwarded to the authorization helper.
options:`dict[str, str] \| None`	Unused; present for API compatibility.
Returns
`Token`	A `Token` containing the access token and expiration.