Skip to content

Remote Source#

RemoteSource is the abstract base for sources that fetch configuration from remote services — secret managers, key-value stores, HTTP APIs. The only implementation shipped today is VaultSource (HashiCorp Vault); the contract is small enough to plug in your own (AWS Secrets Manager, Azure Key Vault, Consul KV, …) by overriding two methods. See Implementing a custom RemoteSource.

Quickstart with VaultSource#

Install the extra (pulls hvac):

pip install dature[vault]              # runtime only
pip install dature[vault,type-stubs]   # runtime + mypy/pyright stubs for hvac
"""Token authentication, KV v2."""

import os
from dataclasses import dataclass

import dature


@dataclass
class Config:
    db_password: str
    port: int
    name: str


config = dature.load(
    dature.VaultSource(
        url=os.environ["VAULT_ADDR"],
        token=os.environ["VAULT_TOKEN"],
        path="myapp/config",
    ),
    schema=Config,
)

assert config == Config(db_password="s3cret", port=5432, name="myapp")  # noqa: S106

VaultSource fields#

  • url — Vault address.
  • path — secret path inside the mount (required).
  • token or role_id + secret_id — authentication (mutually exclusive).
  • mount_point — secrets engine mount; default "secret".
  • kv_version1 or 2; default 2.
  • namespace — Vault Enterprise namespace.
  • verify — TLS verification (True, a CA bundle path, or False).

Global configuration via configure()#

Connection settings rarely change per-call, so they can be set once via dature.configure(vault={...}) (or the matching DATURE_VAULT__* env vars):

"""Set Vault connection settings globally via configure()."""

import os
from dataclasses import dataclass

import dature


@dataclass
class Config:
    db_password: str
    port: int
    name: str


dature.configure(
    vault={
        "url": os.environ["VAULT_ADDR"],
        "token": os.environ["VAULT_TOKEN"],
    },
)

config = dature.load(dature.VaultSource(path="myapp/config"), schema=Config)

assert config == Config(db_password="s3cret", port=5432, name="myapp")  # noqa: S106

Precedence (highest first): instance fields → configure()DATURE_VAULT__* env. None on the instance means "fall through to the next layer". See Configure for the full picture.

Combining with other sources#

RemoteSource composes with file/env/CLI sources via load() like any other source. A common pattern is JSON/YAML for non-sensitive defaults, Vault for secrets, env or CLI for last-mile overrides — order in load() controls precedence (default last_wins). See Merging.

Implementing a custom RemoteSource#

RemoteSource is abstract. To plug in a different backend, subclass it and implement two methods:

  • remote_address() -> str — human-readable identifier shown in error messages and debug reports (e.g. an ARN, a URL, a Consul path).
  • _fetch() -> JSONValue — perform the actual fetch and return a dict (mapping field names to values). The base class handles caching, prefix stripping, env-var expansion, and error-location rendering for free.
"""Custom RemoteSource subclass — pure Python, no external services."""

from dataclasses import dataclass
from typing import ClassVar

import dature
from dature.sources.base import RemoteSource
from dature.types import JSONValue


@dataclass(kw_only=True, repr=False)
class InMemorySource(RemoteSource):
    """Demonstrates the RemoteSource contract: override two methods."""

    backend: dict[str, dict[str, JSONValue]]
    key: str

    format_name = "in-memory"
    location_label: ClassVar[str] = "MEMORY"

    def remote_address(self) -> str:
        return f"memory://{self.key}"

    def _fetch(self) -> JSONValue:
        return self.backend[self.key]


@dataclass
class Config:
    db_password: str
    port: int


backend = {"myapp/config": {"db_password": "s3cret", "port": 5432}}
config = dature.load(
    InMemorySource(backend=backend, key="myapp/config"),
    schema=Config,
)

assert config == Config(db_password="s3cret", port=5432)  # noqa: S106

Optional hooks#

  • validate() — runs after credential merge; override to enforce invariants (e.g. "either token or role_id+secret_id is set", as VaultSource does).
  • __repr__() — defaults to f"{self.format_name} '{self.remote_address()}'".

The config_group ClassVar that ties VaultSource to dature.configure(vault=...) is wired into dature.config.DatureConfig and is not extensible from outside the package. For a custom subclass, expose connection params via constructor arguments.

Custom subclasses are not part of dature's API surface

InMemorySource above is a teaching example