saltext.vault.utils.vault.cache

Vault-specific cache classes

class saltext.vault.utils.vault.cache.CommonCache(context: dict[Any, Any], cbank: str, cache_backend: salt.cache.Cache | None = None, ttl: int | str | None = None, flush_exception: type[VaultConfigExpired] | type[VaultAuthExpired] | None = None)

Base class that unifies context and other cache backends.

class saltext.vault.utils.vault.cache.VaultCache(context: dict[Any, Any], cbank: str, ckey: str, cache_backend: salt.cache.Cache | None = None, ttl: int | str | None = None, flush_exception: type[VaultConfigExpired] | type[VaultAuthExpired] | None = None)

Encapsulates session and other cache backends for a single domain like secret path metadata. Uses a single cache key.

exists(flush: bool = True) → bool

Check whether data for this domain exists

get(flush: bool = True) → dict[str, Any] | None

Return the cached data for this domain or None

flush(cbank: bool = False)

Flush the cache for this domain

store(value: dict[str, Any])

Store data for this domain

class saltext.vault.utils.vault.cache.VaultConfigCache(context: dict[~typing.Any, ~typing.Any], cbank: str, ckey: str, opts: dict[str, ~typing.Any], cache_backend_factory: typing.Callable[[dict[str, typing.Any], dict[typing.Any, typing.Any]], salt.cache.Cache | None] = <function _get_cache_backend>, init_config: dict[str, ~typing.Any] | None = None, flush_exception: type[~saltext.vault.utils.vault.exceptions.VaultConfigExpired] | type[~saltext.vault.utils.vault.exceptions.VaultAuthExpired] | None = None)

Handles caching of received configuration

exists(flush: bool = True) → bool

Check if a configuration has been loaded and cached

get(flush: bool = True) → dict[str, Any] | None

Return the current cached configuration

flush(cbank: bool = True)

Flush all connection-scoped data

store(value: dict[str, Any])

Reload cache configuration, then store the new Vault configuration, overwriting the existing one.

class saltext.vault.utils.vault.cache.LeaseCacheMixin(*args, lease_cls: type[LeaseType], expire_events: Callable[[...], bool] | None = None, **kwargs)

Mixin for auth and lease cache that checks validity and acts with hydrated objects

class saltext.vault.utils.vault.cache.VaultLeaseCache(*args, lease_cls: type[LeaseType], expire_events: Callable[[...], bool] | None = None, **kwargs)

Handles caching of Vault leases. Supports multiple cache keys. Checks whether cached leases are still valid before returning. Does not enforce for per-lease min_ttl.

get(ckey: str, valid_for: int | str = 0, flush: bool = True) → LeaseType | None

Returns valid cached lease data or None. Flushes cache if invalid by default.

store(ckey: str, value: LeaseType | dict[str, Any])

Store a lease in cache

exists(ckey: str, flush: bool = True) → bool

Check whether a named lease exists in cache. Does not filter invalid ones, so fetching a reported one might still return None.

flush(ckey: str | None = None)

Flush the lease cache or a single lease from the lease cache

list() → set[str]

List all cached leases. Does not filter invalid ones, so fetching a reported one might still return None.

class saltext.vault.utils.vault.cache.VaultAuthCache(context: dict[Any, Any], cbank: str, ckey: str, auth_cls: type[AuthType], cache_backend=None, ttl: int | str | None = None, flush_exception: type[VaultConfigExpired] | type[VaultAuthExpired] | None = None)

Implements authentication secret-specific caches. Checks whether the cached secrets are still valid before returning.

exists(flush: bool = True) → bool

Check whether data for this domain exists

get(valid_for: int | str = 0, flush: bool = True) → AuthType | None

Returns valid cached auth data or None. Flushes cache if invalid by default.

store(value: AuthType | dict[str, Any])

Store an auth credential in cache. Overwrites possibly existing one.

flush(cbank: str | None = None)

Flush the cached auth credentials. If this is a token cache, flushing it deletes the whole session-scoped cache bank.