API reference¶

Auto-generated from docstrings via mkdocstrings.

Top-level¶

vaultly — declarative Pydantic-native secrets manager.

Optional backends (AWSSSMBackend, VaultBackend) live in their own submodules and import their SDKs lazily; they are not re-exported here so that import vaultly works without boto3 / hvac installed. Use:

from vaultly.backends.aws_ssm import AWSSSMBackend
from vaultly.backends.vault import VaultBackend

Backend ¶

Bases: ABC

get `abstractmethod` ¶

get(path: str, *, version: int | str | None = None) -> str

Return the raw string at path or raise a vaultly error.

version, if given, pins to a specific version of the secret. Backends that don't support versioning should ignore it.

get_batch ¶

get_batch(paths: list[str]) -> dict[str, str]

Fetch many paths at once (latest versions). Default: serial get.

Duplicates in paths are collapsed before fetching — backends with a real batch API (e.g. SSM GetParameters) reject duplicate names, and a single shared cache key suffices.

RetryingBackend ¶

RetryingBackend(
    inner: Backend,
    *,
    max_attempts: int = 3,
    base_delay: float = 0.5,
    max_delay: float = 4.0,
    total_timeout: float | None = 10.0,
    jitter: bool = True,
    is_retryable: Callable[[BaseException], bool]
    | None = None,
    backoff: Callable[[int], float] | None = None,
    on_retry: Callable[[int, BaseException, float], None]
    | None = None,
    sleep: Callable[[float], None] = time.sleep,
    rng: Callable[[], float] = random.random,
    monotonic: Callable[[], float] = time.monotonic,
)

Bases: Backend

Wrap inner with retry semantics.

Parameters:

Name	Type	Description	Default
`inner`	`Backend`	The backend to delegate to.	required
`max_attempts`	`int`	Cap on total tries (>= 1).	`3`
`base_delay`	`float`	Starting delay for the default exponential schedule.	`0.5`
`max_delay`	`float`	Cap for the default exponential schedule.	`4.0`
`total_timeout`	`float \| None`	Wall-clock budget across all attempts (seconds). `None` disables; otherwise must be > 0.	`10.0`
`jitter`	`bool`	Apply full jitter to the default schedule. Ignored when a custom `backoff=` is given.	`True`
`is_retryable`	`Callable[[BaseException], bool] \| None`	Predicate `(exc) -> bool` deciding whether to retry. Defaults to "retry only TransientError". You'd widen this if your backend reasonably retries other error types (e.g. eventual-consistency `SecretNotFoundError` on a freshly written key).	`None`
`backoff`	`Callable[[int], float] \| None`	Custom delay function `(attempt) -> seconds`. `attempt` is 1-based. When given, replaces the default `base_delay x 2^(attempt-1)`-with-jitter formula entirely.	`None`
`on_retry`	`Callable[[int, BaseException, float], None] \| None`	Callback fired before each retry sleep, with `(attempt, exc, delay)`. Use it to record metrics or push breadcrumbs. Runs synchronously; keep it cheap.	`None`
`sleep`	`Callable[[float], None]`	Override-point for tests; replaces `time.sleep`.	`sleep`
`rng`	`Callable[[], float]`	Override-point for tests; replaces `random.random`.	`random`
`monotonic`	`Callable[[], float]`	Override-point for tests; replaces `time.monotonic`.	`monotonic`

Raises:

Type	Description
`ValueError`	If `max_attempts < 1` or `total_timeout <= 0`.

SecretModel ¶

Bases: BaseModel

refresh ¶

refresh(name: str) -> Any

Invalidate name in the cache and re-fetch from the backend.

Atomic against concurrent fetches: holds the per-key lock across invalidate + fetch so a parallel _fetch can't slip in and repopulate the cache with a stale value between the two steps.

refresh_all ¶

refresh_all() -> None

Invalidate every cached secret in this tree.

prefetch ¶

prefetch() -> None

Eagerly fetch every secret in the tree.

Unversioned secrets are fetched in one backend.get_batch call; versioned secrets fall back to serial get. Safe to call multiple times; uses the root's cache as usual.

AuthError ¶

Bases: VaultlyError

Backend rejected our credentials. Not retried.

ConfigError ¶

Bases: VaultlyError

Problem with the user-provided model configuration.

MissingContextVariableError ¶

Bases: ConfigError

A {var} in a secret path refers to a field that does not exist.

SecretNotFoundError ¶

Bases: VaultlyError

Backend confirmed the secret does not exist. Not retried.

TransientError ¶

Bases: VaultlyError

Transient backend failure (timeout, 5xx, throttling). Retryable.

VaultlyError ¶

Bases: Exception

Base for every exception raised by vaultly.

Secret ¶

Secret(
    path: str,
    *,
    ttl: float | None = None,
    transform: Callable[[str], Any] | None = None,
    version: int | str | None = None,
    description: str | None = None,
) -> Any

Declare a secret-backed field.

Parameters:

Name	Type	Description	Default
`path`	`str`	Backend path. `{var}` placeholders are filled from fields of the root `SecretModel` at fetch time.	required
`ttl`	`float \| None`	Cache lifetime in seconds. `None` = cache forever, `0` = never cache, `>0` = seconds.	`None`
`transform`	`Callable[[str], Any] \| None`	Optional callable applied to the raw backend string, overriding the default type-based cast.	`None`
`version`	`int \| str \| None`	Pin to a specific version of the secret. Backends that don't support versioning ignore this.	`None`
`description`	`str \| None`	Free-text description; surfaces in error messages and in the spec's `repr`. Useful for debugging large models.	`None`

Returns:

Type	Description
`Any`	A Pydantic `FieldInfo` (typed as `Any` so it slots into `field: T = ...`
`Any`	declarations without complaints from type checkers).

`vaultly.SecretModel`¶

vaultly.core.model.SecretModel ¶

Bases: BaseModel

prefetch ¶

prefetch() -> None

Eagerly fetch every secret in the tree.

Unversioned secrets are fetched in one backend.get_batch call; versioned secrets fall back to serial get. Safe to call multiple times; uses the root's cache as usual.

refresh ¶

refresh(name: str) -> Any

Invalidate name in the cache and re-fetch from the backend.

Atomic against concurrent fetches: holds the per-key lock across invalidate + fetch so a parallel _fetch can't slip in and repopulate the cache with a stale value between the two steps.

refresh_all ¶

refresh_all() -> None

Invalidate every cached secret in this tree.

`vaultly.Secret`¶

vaultly.core.secret.Secret ¶

Secret(
    path: str,
    *,
    ttl: float | None = None,
    transform: Callable[[str], Any] | None = None,
    version: int | str | None = None,
    description: str | None = None,
) -> Any

Declare a secret-backed field.

Parameters:

Name	Type	Description	Default
`path`	`str`	Backend path. `{var}` placeholders are filled from fields of the root `SecretModel` at fetch time.	required
`ttl`	`float \| None`	Cache lifetime in seconds. `None` = cache forever, `0` = never cache, `>0` = seconds.	`None`
`transform`	`Callable[[str], Any] \| None`	Optional callable applied to the raw backend string, overriding the default type-based cast.	`None`
`version`	`int \| str \| None`	Pin to a specific version of the secret. Backends that don't support versioning ignore this.	`None`
`description`	`str \| None`	Free-text description; surfaces in error messages and in the spec's `repr`. Useful for debugging large models.	`None`

Returns:

Type	Description
`Any`	A Pydantic `FieldInfo` (typed as `Any` so it slots into `field: T = ...`
`Any`	declarations without complaints from type checkers).

Backends¶

vaultly.Backend ¶

get `abstractmethod` ¶

get(path: str, *, version: int | str | None = None) -> str

Return the raw string at path or raise a vaultly error.

version, if given, pins to a specific version of the secret. Backends that don't support versioning should ignore it.

get_batch ¶

get_batch(paths: list[str]) -> dict[str, str]

Fetch many paths at once (latest versions). Default: serial get.

Duplicates in paths are collapsed before fetching — backends with a real batch API (e.g. SSM GetParameters) reject duplicate names, and a single shared cache key suffices.

vaultly.EnvBackend ¶

EnvBackend(prefix: str = '')

Bases: Backend

vaultly.MockBackend ¶

MockBackend(
    data: dict[str, str] | None = None,
    *,
    versions: dict[tuple[str, int | str], str]
    | None = None,
)

Bases: Backend

vaultly.RetryingBackend ¶

RetryingBackend(
    inner: Backend,
    *,
    max_attempts: int = 3,
    base_delay: float = 0.5,
    max_delay: float = 4.0,
    total_timeout: float | None = 10.0,
    jitter: bool = True,
    is_retryable: Callable[[BaseException], bool]
    | None = None,
    backoff: Callable[[int], float] | None = None,
    on_retry: Callable[[int, BaseException, float], None]
    | None = None,
    sleep: Callable[[float], None] = time.sleep,
    rng: Callable[[], float] = random.random,
    monotonic: Callable[[], float] = time.monotonic,
)

Bases: Backend

Wrap inner with retry semantics.

Parameters:

Name	Type	Description	Default
`inner`	`Backend`	The backend to delegate to.	required
`max_attempts`	`int`	Cap on total tries (>= 1).	`3`
`base_delay`	`float`	Starting delay for the default exponential schedule.	`0.5`
`max_delay`	`float`	Cap for the default exponential schedule.	`4.0`
`total_timeout`	`float \| None`	Wall-clock budget across all attempts (seconds). `None` disables; otherwise must be > 0.	`10.0`
`jitter`	`bool`	Apply full jitter to the default schedule. Ignored when a custom `backoff=` is given.	`True`
`is_retryable`	`Callable[[BaseException], bool] \| None`	Predicate `(exc) -> bool` deciding whether to retry. Defaults to "retry only TransientError". You'd widen this if your backend reasonably retries other error types (e.g. eventual-consistency `SecretNotFoundError` on a freshly written key).	`None`
`backoff`	`Callable[[int], float] \| None`	Custom delay function `(attempt) -> seconds`. `attempt` is 1-based. When given, replaces the default `base_delay x 2^(attempt-1)`-with-jitter formula entirely.	`None`
`on_retry`	`Callable[[int, BaseException, float], None] \| None`	Callback fired before each retry sleep, with `(attempt, exc, delay)`. Use it to record metrics or push breadcrumbs. Runs synchronously; keep it cheap.	`None`
`sleep`	`Callable[[float], None]`	Override-point for tests; replaces `time.sleep`.	`sleep`
`rng`	`Callable[[], float]`	Override-point for tests; replaces `random.random`.	`random`
`monotonic`	`Callable[[], float]`	Override-point for tests; replaces `time.monotonic`.	`monotonic`

Raises:

Type	Description
`ValueError`	If `max_attempts < 1` or `total_timeout <= 0`.

Cloud backends¶

These require their respective extras (pip install 'vaultly[aws]' / 'vaultly[vault]').

vaultly.backends.aws_ssm.AWSSSMBackend ¶

AWSSSMBackend(
    *,
    region_name: str | None = None,
    client: Any = None,
    with_decryption: bool = True,
    config: Config | None = None,
)

Bases: Backend

vaultly.backends.vault.VaultBackend ¶

VaultBackend(
    *,
    url: str | None = None,
    token: str | None = None,
    mount_point: str = "secret",
    default_key: str = "value",
    client: Any = None,
    token_factory: Callable[[], str] | None = None,
    reuse_connection: bool = True,
    idle_timeout: float | None = None,
    client_kwargs: dict[str, Any] | None = None,
)

Bases: Backend

Initialize the Vault backend.

Parameters:

Name	Type	Description	Default
`url`	`str \| None`	Vault server URL.	`None`
`token`	`str \| None`	Static auth token.	`None`
`mount_point`	`str`	KV v2 mount point (default `"secret"`).	`'secret'`
`default_key`	`str`	Field within the secret dict to return when the path doesn't carry a `:key` suffix.	`'value'`
`client`	`Any`	Pass a pre-configured `hvac.Client` instead of letting us create one. When given, `reuse_connection`, `idle_timeout`, and `client_kwargs` are ignored — you own the client's lifecycle.	`None`
`token_factory`	`Callable[[], str] \| None`	Optional callable invoked when the current token is rejected (`Unauthorized`). Should return a fresh token; we install it on the client and retry the read once. Use this with short-lived tokens (AppRole, K8s auth, etc.).	`None`
`reuse_connection`	`bool`	When `True` (default), one persistent `hvac.Client` (and its underlying `requests.Session`) is kept for the backend's lifetime. Set `False` for a fresh client per call — useful when reads are infrequent and idle TCP connections get dropped by an LB / proxy.	`True`
`idle_timeout`	`float \| None`	Recreate the persistent client when the gap between calls exceeds this many seconds. Compromise between `reuse_connection=True` (cheap, may stale) and `reuse_connection=False` (always fresh, slow). Ignored if `reuse_connection=False`.	`None`
`client_kwargs`	`dict[str, Any] \| None`	Extra kwargs forwarded to `hvac.Client(...)` (e.g. `verify=False`, custom `adapter`). Used when we construct or recreate the client; ignored when `client=` was passed.	`None`

Errors¶

vaultly.VaultlyError ¶

Bases: Exception

Base for every exception raised by vaultly.

vaultly.ConfigError ¶

Bases: VaultlyError

Problem with the user-provided model configuration.

vaultly.MissingContextVariableError ¶

Bases: ConfigError

A {var} in a secret path refers to a field that does not exist.

vaultly.SecretNotFoundError ¶

Bases: VaultlyError

Backend confirmed the secret does not exist. Not retried.

vaultly.AuthError ¶

Bases: VaultlyError

Backend rejected our credentials. Not retried.

vaultly.TransientError ¶

Bases: VaultlyError

Transient backend failure (timeout, 5xx, throttling). Retryable.

API reference¶

Top-level¶

Backend ¶

get abstractmethod ¶

get_batch ¶

RetryingBackend ¶

SecretModel ¶

refresh ¶

refresh_all ¶

prefetch ¶

AuthError ¶

ConfigError ¶

MissingContextVariableError ¶

SecretNotFoundError ¶

TransientError ¶

VaultlyError ¶

Secret ¶

vaultly.SecretModel¶

vaultly.core.model.SecretModel ¶

prefetch ¶

refresh ¶

refresh_all ¶

vaultly.Secret¶

vaultly.core.secret.Secret ¶

Backends¶

vaultly.Backend ¶

get abstractmethod ¶

get_batch ¶

vaultly.EnvBackend ¶

vaultly.MockBackend ¶

vaultly.RetryingBackend ¶

Cloud backends¶

vaultly.backends.aws_ssm.AWSSSMBackend ¶

vaultly.backends.vault.VaultBackend ¶

Errors¶

vaultly.VaultlyError ¶

vaultly.ConfigError ¶

vaultly.MissingContextVariableError ¶

vaultly.SecretNotFoundError ¶

vaultly.AuthError ¶

vaultly.TransientError ¶

get `abstractmethod` ¶

`vaultly.SecretModel`¶

`vaultly.Secret`¶

get `abstractmethod` ¶