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.

Method __init__ Undocumented
Async Method get Get the cached token, respecting the throttle.
Method get_cached Get the cached token without checking the throttle.
Async Method refresh Refresh the cached token by fetching it again.
Async Method remove Remove the token from the cache.
Async Method remove_if_equal Remove the token from the cache if it matches the provided token.
Async Method set Set the token in the cache and update the throttle.
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)

Undocumented

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

Get the cached token, respecting the throttle.

def get_cached(self) -> Token | None: (source)

Get the cached token without checking the throttle.

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

Refresh the cached token by fetching it again.

async def remove(self): (source)

Remove the token from the cache.

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

Remove the token from the cache if it matches the provided token.

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

Set the token in the cache and update the throttle.

_cache = (source)

Undocumented

_cached_token: Token | None = (source)

Undocumented

_name = (source)

Undocumented

_next_check: datetime = (source)

Undocumented

_throttle: timedelta = (source)

Undocumented