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).tokenorrole_id+secret_id— authentication (mutually exclusive).mount_point— secrets engine mount; default"secret".kv_version—1or2; default2.namespace— Vault Enterprise namespace.verify— TLS verification (True, a CA bundle path, orFalse).
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. "eithertokenorrole_id+secret_idis set", asVaultSourcedoes).__repr__()— defaults tof"{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