API Reference¶
Root Exports¶
from vault_kv_client import (
SecretNotFoundError,
VaultAuth,
VaultClientError,
VaultDependencyError,
VaultJWTAuth,
VaultKubernetesAuth,
VaultManager,
VaultNotConfiguredError,
VaultSettings,
get_creds,
get_default_manager,
)
Models¶
VaultSettings¶
addr: strverify: bool | str = Truenamespace: str | None = None
VaultAuth¶
Exactly one of the following must be set:
tokenapprolekubernetesjwt
VaultKubernetesAuth¶
role: strmount_point: str = "k8s"jwt: str | None = Nonejwt_file: str = "/var/run/secrets/kubernetes.io/serviceaccount/token"
VaultJWTAuth¶
role: str | None = Nonemount_point: str = "jwt"jwt: str | None = Nonejwt_file: str | None = None
VaultManager¶
get_secret(mount_point, path, kv_version=None, raise_on_deleted_version=True) -> dict¶
Read and normalize a secret payload.
upsert_secret(mount_point=..., path=..., secret=..., kv_version=None) -> None¶
Create or update a secret payload.
list_secrets(mount_point, path="", kv_version=None) -> list[str]¶
List direct child keys below the requested prefix.
list_all_secrets(mount_point, path="", kv_version=None) -> list[str]¶
Recursively collect all secret paths below the requested prefix.
copy_secret(source_mount=..., target_mount=..., path=..., source_kv_version=None, target_kv_version=None) -> None¶
Copy a secret between mount points.
clear_cache() -> None¶
Invalidate the in-memory read cache.
Convenience Helpers¶
get_default_manager() -> VaultManager¶
Create a lazily initialized manager from environment variables and optional Airflow Variable fallbacks.
get_creds(mount_point, path, **kwargs) -> dict¶
Thin wrapper around get_default_manager().get_secret(...).