nebius.aio.token.file_cache.renewable_bearer.RenewableFileCacheBearer

class documentation

class RenewableFileCacheBearer(ParentBearer): (source)

Constructor: RenewableFileCacheBearer(bearer, safety_margin, cache_file, throttle)

Bearer that composes a wrapped bearer with a file-backed cache.

Example

Wrap a custom bearer with a name and file cache:

from nebius.sdk import SDK
from nebius.aio.token.token import Bearer, Receiver, Token
from nebius.aio.token.file_cache.renewable_bearer import (
    RenewableFileCacheBearer
)

class SomeCustomHeavyLoadBearer(Bearer):
    def receiver(self) -> Receiver:
        return SomeReceiver()

class SomeReceiver(Receiver):
    async def _fetch(self, timeout=None, options=None) -> Token:
        # Simulate heavy load token fetch
        return Token("heavy-token")

    def can_retry(self, err, options=None) -> bool:
        return False

custom_bearer = SomeCustomHeavyLoadBearer()
cached_bearer = RenewableFileCacheBearer(custom_bearer)

sdk = SDK(credentials=cached_bearer)

Parameters
bearer	Wrapped bearer used for refreshing tokens. Must be named.
safety_margin	Safety margin before token expiration.
cache_file	Path to the file used for persistent cache.
throttle	In-memory throttle interval for cache reads.

Method	`__init__`	Create a renewable file-backed bearer.
Method	`receiver`	Return a `RenewableFileCacheReceiver` bound to the cache.
Instance Variable	`safety_margin`	A `datetime.timedelta` or seconds value used to determine when a token is considered too close to expiry. If `None`, the cache will be used until tokens are expired.
Property	`wrapped`	Return the wrapped bearer instance.
Instance Variable	`_bearer`	Undocumented
Instance Variable	`_cache`	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, bearer: ParentBearer, safety_margin: timedelta | float = timedelta(hours=2), cache_file: str | Path = Path(DEFAULT_CONFIG_DIR) / DEFAULT_CREDENTIALS_FILE, throttle: timedelta | float = timedelta(minutes=5)): (source) ¶

Create a renewable file-backed bearer.

The constructor creates the underlying ThrottledTokenCache.

Raises
`ValueError`	When the wrapped bearer has no name.

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

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

Return a RenewableFileCacheReceiver bound to the cache.

Returns
`ParentReceiver`	A receiver that uses the file cache and refreshes from the wrapped bearer when necessary.

safety_margin: timedelta | None = (source) ¶

A datetime.timedelta or seconds value used to determine when a token is considered too close to expiry. If None, the cache will be used until tokens are expired.

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

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

Return the wrapped bearer instance.

Returns
The wrapped bearer passed at construction.

_bearer = (source) ¶

Undocumented

_cache = (source) ¶

Undocumented