nebius.aio.token.renewable

Notes

The renewal bearer starts a background task on the first token fetch and keeps the cached token refreshed up to a configurable safe fraction of the token lifetime. When a fetch request requires a renewed token it may either wait for the renewal to complete (synchronous mode) or trigger a background renewal and wait for the cached token to become fresh (asynchronous mode).

Examples

Basic usage (asynchronous renewal):

from nebius.aio.token.static import Bearer as StaticBearer
from nebius.aio.token.renewable import Bearer as RenewableBearer
import asyncio

async def demo():
    src = StaticBearer("my-static-token")
    bearer = RenewableBearer(src)
    receiver = bearer.receiver()
    tok = await receiver.fetch(timeout=5)
    print(tok)

asyncio.run(demo())

Synchronous renewal request (waits for a fresh token, raising on timeout):

from nebius.aio.token.static import Bearer as StaticBearer
from nebius.aio.token.renewable import Bearer as RenewableBearer
from nebius.aio.token.renewable import (
    OPTION_RENEW_SYNCHRONOUS,
    OPTION_RENEW_REQUEST_TIMEOUT,
)
import asyncio

async def sync_demo():
    src = StaticBearer("my-static-token")
    bearer = RenewableBearer(src)
    receiver = bearer.receiver()
    options = {
        OPTION_RENEW_SYNCHRONOUS: "1",
        OPTION_RENEW_REQUEST_TIMEOUT: "2.0",
    }
    tok = await receiver.fetch(timeout=5, options=options)
    print(tok)

asyncio.run(sync_demo())