Skip to content

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: str
  • verify: bool | str = True
  • namespace: str | None = None

VaultAuth

Exactly one of the following must be set:

  • token
  • approle
  • kubernetes
  • jwt

VaultKubernetesAuth

  • role: str
  • mount_point: str = "k8s"
  • jwt: str | None = None
  • jwt_file: str = "/var/run/secrets/kubernetes.io/serviceaccount/token"

VaultJWTAuth

  • role: str | None = None
  • mount_point: str = "jwt"
  • jwt: str | None = None
  • jwt_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(...).