Skip to content
GitHub

API Reference

Protocols

Section titled “Protocols”

Async persistent store for sensitive named secret values.

Typical usage

store = await container.resolve(AsyncSecretStoreProtocol)
token = await store.get("stripe_api_key")
store = await container.resolve(AsyncSecretStoreProtocol)
token = await store.get("stripe_api_key")
get
async def get(name: str) -> str | None

Return the secret value for name, or None if absent.

Parameters
ParameterTypeDescription
`name`strUnique secret identifier.
get_bulk
async def get_bulk(*names: str) -> dict[str, str]

Return a mapping of name → value for all requested secrets.

Parameters
ParameterTypeDescription
`names`strOne or more secret names.
Returns
TypeDescription
dict[str, str]Dict containing only the names that were found.
set
async def set(
    name: str,
    value: str
) -> None

Write or overwrite a secret value.

Parameters
ParameterTypeDescription
`name`strUnique secret identifier.
`value`strPlaintext secret value.
delete
async def delete(name: str) -> None

Remove a secret. No-op if absent.

Parameters
ParameterTypeDescription
`name`strUnique secret identifier.

An ``AsyncSecretStoreProtocol`` that also supports credential rotation and versioned history.
rotate
async def rotate(key: str) -> VersionedSecret

Generate a new value for key, store it as the next version, and return the resulting VersionedSecret.

get_version
async def get_version(
    key: str,
    version: int
) -> str | None

Return the value of a specific version, or None if absent.

list_versions
async def list_versions(key: str) -> list[SecretVersion]

Return metadata for every known version of key, newest first.

get_current_version
async def get_current_version(key: str) -> VersionedSecret

Return the full VersionedSecret for the current (latest) version. Raises KeyError if key does not exist.

Classes

Section titled “Classes”

Wraps a ``RotatableSecretStoreProtocol`` with age-checked automatic rotation.

On get_rotated the decorator checks the current version’s age against schedule.max_age_seconds. If exceeded it calls rotate on the underlying store automatically.

After rotation the old version is still served via get_current_version until the grace period (grace_period_seconds) elapses. The VersionedSecret returned after rotation has its expires_at field populated.

__init__
def __init__(
    store: RotatableSecretStoreProtocol,
    schedule: RotationSchedule
) -> None
get_rotated
async def get_rotated(key: str) -> tuple[str, bool]

Return (value, was_rotated).

If the current version of key is older than max_age_seconds a new version is generated first. The old version is preserved in a grace buffer so that callers still see it until the grace period expires.

get_current_version
async def get_current_version(key: str) -> VersionedSecret

Return the current version, respecting the grace period.

If key was recently rotated and the grace period has not yet elapsed the old version is returned. Once the grace period expires the new version from the underlying store is served.

check_warnings
async def check_warnings(key: str) -> str | None

Return a warning message if key is approaching its rotation deadline, or None if no warning is needed.

Policy for automatic rotation of credentials.

A secret value was read.

Payload fired when a secret value is read.

Attributes: key: The secret key that was accessed.

Wraps a ``RotatableSecretStoreProtocol``, logging all operations via an ``AuditLoggerProtocol``.

Every read, write, delete, and rotation produces an AuditEntry with the action, the secret name, and a timestamp.

__init__
def __init__(
    store: RotatableSecretStoreProtocol,
    logger: AuditLoggerProtocol,
    actor_id: str = 'secrets-system',
    context: Context | None = None
) -> None
get
async def get(name: str) -> str | None
get_bulk
async def get_bulk(*names: str) -> dict[str, str]
set
async def set(
    name: str,
    value: str
) -> None
delete
async def delete(name: str) -> None
rotate
async def rotate(key: str) -> VersionedSecret
get_version
async def get_version(
    key: str,
    version: int
) -> str | None
list_versions
async def list_versions(key: str) -> list[SecretVersion]
get_current_version
async def get_current_version(key: str) -> VersionedSecret

A new secret was stored.

Payload fired when a new secret is stored.

Attributes: key: The secret key that was created.

A secret was deleted.

Payload fired when a secret is deleted.

Attributes: key: The secret key that was deleted.

A secret was rotated to a new version.

Payload fired when a secret is rotated to a new version.

Attributes: key: The secret key that was rotated. new_version: The version number after rotation.

Lightweight metadata for a single version of a secret.

Top-level configuration for the secrets rotation subsystem.

Attributes: name: Configuration name. enabled: Whether the secrets subsystem is enabled. backend_type: Backend store type ("memory" or "vault"). backend_options: Keyword arguments forwarded to the backend constructor. max_age_seconds: Maximum age before automatic rotation. warning_before_seconds: Seconds before expiry to emit rotation warnings. tenant_id: Optional tenant namespace for multi-tenancy. audit_actor_id: Actor identifier for audit log entries.

is_production
property is_production() -> bool
is_development
property is_development() -> bool
is_test
property is_test() -> bool

IoC module for the secrets/credential vault subsystem.
configure
def configure(
    cls,
    config: SecretsConfig | None = None
) -> DynamicModule
stub
def stub(
    cls,
    config: SecretsConfig | None = None
) -> DynamicModule

Return a DynamicModule pre-configured for testing.

Wraps a ``RotatableSecretStoreProtocol``, prefixing all secret names with ``tenant_id/`` to isolate secrets per tenant.

Example

tenant_a_store = TenantScopedSecretStore(store, "tenant_a")
await tenant_a_store.set("db_password", "s3cret")
# internally stored as "tenant_a/db_password"
tenant_a_store = TenantScopedSecretStore(store, "tenant_a")
await tenant_a_store.set("db_password", "s3cret")
# internally stored as "tenant_a/db_password"
__init__
def __init__(
    store: RotatableSecretStoreProtocol,
    tenant_id: str
) -> None
get
async def get(name: str) -> str | None
get_bulk
async def get_bulk(*names: str) -> dict[str, str]
set
async def set(
    name: str,
    value: str
) -> None
delete
async def delete(name: str) -> None
rotate
async def rotate(key: str) -> VersionedSecret
get_version
async def get_version(
    key: str,
    version: int
) -> str | None
list_versions
async def list_versions(key: str) -> list[SecretVersion]
get_current_version
async def get_current_version(key: str) -> VersionedSecret

A secret value together with its version metadata.

Exceptions

Section titled “Exceptions”

Raised when access to a secret is denied.

Raised when the underlying secret backend (e.g. Vault) fails.

Raised when secrets subsystem configuration is invalid.

Raised when a requested secret does not exist.
__init__
def __init__(
    key: str,
    **kwargs: Any
) -> None

Raised when automatic rotation of a secret fails.
__init__
def __init__(
    key: str,
    reason: str = '',
    **kwargs: Any
) -> None

Base exception for all secrets-domain errors.