nebius.aio.token.file_cache.throttled_token_cache.ThrottledTokenCache

class documentation

class ThrottledTokenCache: (source)

Constructor: ThrottledTokenCache(name, cache_file, throttle)

View In Hierarchy

A throttled file-based token cache.

The helper keeps an in-memory cached token for throttle seconds to avoid repeated reads from the underlying TokenCache.

Parameters
nameThe key name under which the token is stored in the underlying TokenCache.
cache_filePath to the underlying YAML cache file.
throttleDuration (timedelta or seconds as float) to hold the in-memory cached token. If a float is supplied it is interpreted as seconds.
Method __init__ Create a throttled token cache.
Async Method get Return the cached token, refreshing from disk if throttling allows.
Method get_cached Return the in-memory cached token without consulting disk.
Async Method refresh Refresh the in-memory token from the underlying cache.
Async Method remove Remove the token from both the underlying cache and the in-memory cache.
Async Method remove_if_equal Remove the token if it equals the provided token.
Async Method set Store the token in the underlying cache and update the in-memory cache and throttle expiration.
Instance Variable _cache Undocumented
Instance Variable _cached_token Undocumented
Instance Variable _name Undocumented
Instance Variable _next_check Undocumented
Instance Variable _throttle Undocumented
def __init__(self, name: str, cache_file: str | Path = Path(DEFAULT_CONFIG_DIR) / DEFAULT_CREDENTIALS_FILE, throttle: timedelta | float = timedelta(minutes=5)): (source)

Create a throttled token cache.

async def get(self) -> Token | None: (source)

Return the cached token, refreshing from disk if throttling allows.

If the in-memory token is present, not expired and the throttle period has not elapsed the cached token is returned. Otherwise the underlying TokenCache.get is called.

Returns
Token | NoneThe cached Token or None if no valid token is found.
def get_cached(self) -> Token | None: (source)

Return the in-memory cached token without consulting disk.

Returns
Token | NoneThe cached Token or None if no token is cached.
async def refresh(self) -> Token | None: (source)

Refresh the in-memory token from the underlying cache.

If the token exists on disk and is not expired the in-memory cache and throttle are updated and the token is returned.

async def remove(self): (source)

Remove the token from both the underlying cache and the in-memory cache.

async def remove_if_equal(self, token: Token): (source)

Remove the token if it equals the provided token.

This performs an atomic check-and-remove against the underlying cache and updates the in-memory state when the removal occurs.

async def set(self, token: Token): (source)

Store the token in the underlying cache and update the in-memory cache and throttle expiration.

If the provided token equals the in-memory cached token no write is performed.

_cache = (source)

Undocumented

_cached_token: Token | None = (source)

Undocumented

_name = (source)

Undocumented

_next_check: datetime = (source)

Undocumented

_throttle: timedelta = (source)

Undocumented