nebius.aio.token.file_cache.token

class documentation

class TokenCache: (source)

Constructor: TokenCache(cache_file, path_create_mode, file_create_mode, flock_timeout)

A simple asynchronous file-based token cache.

The cache stores named tokens in a YAML document on disk and uses a file lock to serialize concurrent access. Tokens are automatically filtered for expiration when reading and writing.

Parameters
cache_file	Filesystem path to the YAML cache file. When a string is supplied it is expanded with `pathlib.Path.expanduser`.
path_create_mode	Mode used when creating parent directories (default 0o750).
file_create_mode	Mode used when creating the cache file (default 0o600).
flock_timeout	Timeout in seconds to wait for the file lock when accessing the cache. `None` means wait indefinitely.

Method	`__init__`	Create or reference a token cache file.
Async Method	`get`	Return the token with the given name or `None`.
Async Method	`remove`	Remove the token with the specified name from the cache.
Async Method	`remove_if_equal`	Remove the token only if its value equals the provided token.
Async Method	`set`	Store a token in the cache under the given name.
Instance Variable	`cache_file`	Path to the YAML cache file.
Instance Variable	`file_create_mode`	Mode used when creating the cache file.
Instance Variable	`flock_timeout`	Timeout in seconds to wait for the file lock. `None` means wait indefinitely.
Instance Variable	`path_create_mode`	Mode used when creating parent directories.
Method	`_yaml_dump`	Serialize tokens to a YAML document.
Method	`_yaml_parse`	Parse YAML content and return a mapping of token name -> Token.

def __init__(self, cache_file: str | Path = Path(DEFAULT_CONFIG_DIR) / DEFAULT_CREDENTIALS_FILE, path_create_mode: int = 488, file_create_mode: int = 384, flock_timeout: float | None = 5.0): (source) ¶

Create or reference a token cache file.

async def get(self, name: str) -> Token | None: (source) ¶

Return the token with the given name or None.

The function opens the cache file under a shared/read lock and parses the YAML content. If the token is present and not expired it is returned. Expired tokens are removed from the cache.

Parameters
name:`str`	Token name to lookup.
Returns
`Token \| None`	`Token` instance or `None` when not found or expired.

async def remove(self, name: str): (source) ¶

Remove the token with the specified name from the cache.

The function opens the cache under an exclusive lock, removes the entry when present and writes the updated YAML document back to disk.

Parameters
name:`str`	Token name to remove.
Raises
`ValueError`	If the cache cannot be parsed or written.

async def remove_if_equal(self, name: str, token: Token): (source) ¶

Remove the token only if its value equals the provided token.

The comparison uses Token.__eq__.

Parameters
name:`str`	Token name to remove.
token:`Token`	Token value to compare against.
Raises
`ValueError`	If the cache cannot be parsed or written.

async def set(self, name: str, token: Token): (source) ¶

Store a token in the cache under the given name.

The cache file and its parent directory are created with the configured permissions if necessary. The function acquires an exclusive lock while reading and writing the YAML document.

Parameters
name:`str`	Name under which to store the token.
token:`Token`	`Token` instance to store.
Raises
`ValueError`	If the token cannot be serialized or written.

cache_file: str or pathlib.Path = (source) ¶

Path to the YAML cache file.

file_create_mode = (source) ¶

Mode used when creating the cache file.

flock_timeout = (source) ¶

Timeout in seconds to wait for the file lock. None means wait indefinitely.

path_create_mode = (source) ¶

Mode used when creating parent directories.

def _yaml_dump(self, tokens: dict[str, Token]) -> str: (source) ¶

Serialize tokens to a YAML document.

Expired tokens are omitted from the serialized output.

Parameters
tokens:`dict[str, Token]`	Mapping of token name to `Token`.
Returns
`str`	YAML string or an empty string when there are no tokens to persist.

def _yaml_parse(self, data: str) -> dict[str, Token]: (source) ¶

Parse YAML content and return a mapping of token name -> Token.

The function expects a YAML mapping with a top-level tokens key containing a mapping of string names to token dictionaries produced by Token.to_dict.

Parameters
data:`str`	YAML document as a string.
Returns
`dict[str, Token]`	Mapping from token name to `Token`.
Raises
`ValueError`	When the YAML is invalid or has an unexpected structure.

nebius.aio.token.file_cache.token_cache.TokenCache

`nebius.aio.token.file_cache.token_cache.TokenCache`