API Reference#

Core#

dature.load()#

def load(  # noqa: PLR0913
    *sources: SourceProtocol,
    schema: type[Any] | None = None,
    cache: bool | timedelta | None = None,
    cache_engine: bool | None = None,
    debug: bool | None = None,
    strategy: MergeStrategyName | SourceMergeStrategy = _DEFAULT_STRATEGY,
    field_merges: FieldMergeMap | None = None,
    field_groups: Sequence[FieldGroupTuple] = (),
    root_validators: Iterable[RootPredicate] = (),
    skip_if_broken: bool = False,
    skip_if_missing: bool = False,
    skip_field_if_invalid: SkipFieldsInvalid = None,
    skip_invalid_fields: Any = UNSET,
    expand_env_vars: ExpandEnvVarsMode | None = None,
    secret_field_names: Sequence[str] | None = None,
    mask_secrets: bool | None = None,
    type_loaders: TypeLoaderMap | None = None,
    nested_resolve_strategy: NestedResolveStrategy | None = None,
    nested_resolve: NestedResolve | None = None,
) -> Any:

Main entry point. Two calling patterns:

Function mode — pass schema, get an instance back:

from dataclasses import dataclass

import dature


@dataclass
class Config:
    host: str
    port: int
    debug: bool = False


config = dature.load(
    dature.Yaml12Source(file=SHARED_DIR / "common_app.yaml"),
    schema=Config,
)

assert config.host == "localhost"
assert config.port == 8080
assert config.debug is False

Decorator mode — omit schema, get a decorator:

from dataclasses import dataclass

import dature


@dature.load(dature.Yaml12Source(file=SHARED_DIR / "common_app.yaml"))
@dataclass
class Config:
    host: str
    port: int
    debug: bool = False


config = Config()

assert config.host == "localhost"
assert config.port == 8080
assert config.debug is False

Parameters:

Returns:

• Function mode (schema provided): an instance of schema populated from the sources.
• Decorator mode (schema=None): a decorator that adds load() logic to the decorated dataclass.

Raises:

• TypeError — no sources passed, or a positional argument is not a Source instance.
• DatureConfigError — aggregated field loading errors.
• MergeConflictError — conflicting values with strategy="raise_on_conflict".
• FieldGroupError — field group constraint violation.
• EnvVarExpandError — missing env vars with expand_env_vars="strict".

dature.Loader#

class Loader[T: DataclassInstance]:
    def __init__(*sources, schema, cache=None, cache_engine=None, debug=None, **load_kwargs): ...
    def load(self) -> T: ...

Public class that carries all the load-time parameters and the cache state. Use it for function-mode caching across repeated calls (the throwaway Loader constructed inside dature.load(...) cannot cache between calls). Constructor accepts the same parameters as dature.load(..., schema=...) for function mode. See Caching for the cache semantics (eternal / TTL / bucket-aligned).

Source#

@dataclass(kw_only=True, repr=False)
class Source(abc.ABC):
    prefix: "DotSeparatedPath | None" = None
    name_style: "NameStyle | None" = None
    field_mapping: "FieldMapping | None" = None
    validators: "FieldValidators | None" = None
    expand_env_vars: "ExpandEnvVarsMode | None" = None
    skip_field_if_invalid: "SkipFieldsInvalid" = None
    type_loaders: "TypeLoaderMap | None" = None
    tag: str | None = None
    when: "Condition | None" = None

    format_name: str = ""
    location_label: str = ""
    config_group: str | None = None

Abstract base class for all sources. See Introduction — Source Reference for parameter descriptions.

Parameters:

Public methods:

FileSource(Source)#

Base class for file-based sources (JsonSource, Yaml11Source, Toml10Source, IniSource, etc.).

@dataclass(kw_only=True, repr=False)
class FileFieldMixin:
    file: "FileLike | FilePath | None" = None
    search_system_paths: bool | None = None
    system_config_dirs: "SystemConfigDirsArg | None" = None
    encoding: str | None = None
    skip_if_broken: bool | None = None
    skip_if_missing: bool | None = None
    resolved_file_path: Path | None = field(init=False, default=None)

Overridden methods:

FlatKeySource(Source)#

Base class for flat key=value sources (EnvSource, EnvFileSource, DockerSecretsSource).

@dataclass(kw_only=True, repr=False)
class FlatKeySource(Source, abc.ABC):
    nested_sep: str = "__"
    nested_resolve_strategy: "NestedResolveStrategy | None" = None
    nested_resolve: NestedResolve | None = None

Behavior: All values are strings. Automatic parsing of str, float, date, datetime, time, bytearray, bool, None, str | None. Nested JSON in values ([...], {...}) is inferred. load_raw() returns LoadRawResult with nested_conflicts populated when both flat and JSON keys exist for the same field.

Merge Strategies#

Strategies for resolving field values across multiple sources. Set via strategy parameter of load().

The built-ins are also exposed as classes from dature.strategies.source (SourceLastWins, SourceFirstWins, SourceFirstFound, SourceRaiseOnConflict) implementing the public SourceMergeStrategy Protocol. Pass any object satisfying that protocol as strategy for custom merge logic — see Custom Source Strategy.

Field Merge Strategies#

Per-field overrides via field_merges parameter. Maps F[Config].field to a strategy name, a plain Callable[[list[JSONValue]], JSONValue], or any object implementing the public FieldMergeStrategy Protocol.

The built-ins are also exposed as classes from dature.strategies.field (FieldFirstWins, FieldLastWins, FieldAppend, FieldAppendUnique, FieldPrepend, FieldPrependUnique). See Custom Field Strategy for examples.

Field Path#

F#

Factory for building type-safe field paths. Used for field_mapping, field_merges, field_groups, validators, skip_field_if_invalid, and nested_resolve.

from dataclasses import dataclass

import dature


@dataclass
class Database:
    host: str
    port: int


@dataclass
class Config:
    database: Database
    host: str


path_eager = dature.F[Config].host
path_nested = dature.F[Config].database.host
path_string = dature.F["Config"].host

FieldPath#

Immutable dataclass (frozen=True, slots=True) created via F[Config].field_name.

Methods:

Report#

load_report()#

def load_report(instance: Any) -> LoadReport | None:  # noqa: ANN401
    report = getattr(instance, _REPORT_ATTR, None)
    if isinstance(report, LoadReport):
        return report
    warnings.warn(
        "To get LoadReport, pass debug=True to load()",
        stacklevel=2,
    )
    return None

Retrieves the LoadReport attached to a loaded instance. Returns None and emits a warning if debug=True was not passed to load().

LoadReport, SourceEntry, FieldOrigin#

@dataclass(frozen=True, slots=True, kw_only=True)
class SourceEntry:
    index: int
    file_path: str | None
    loader_type: str
    raw_data: JSONValue


@dataclass(frozen=True, slots=True, kw_only=True)
class FieldOrigin:
    key: str
    value: JSONValue
    source_index: int
    source_file: str | None
    source_loader_type: str


@dataclass(frozen=True, slots=True, kw_only=True)
class LoadReport:
    dataclass_name: str
    strategy: SourceMergeStrategy | None
    sources: tuple[SourceEntry, ...]
    field_origins: tuple[FieldOrigin, ...]
    merged_data: JSONValue

SourceEntry#

Frozen dataclass describing one source in the load pipeline.

FieldOrigin#

Frozen dataclass describing which source provided a specific field value.

LoadReport#

Frozen dataclass with full load diagnostics.

Configuration#

configure()#

def configure(
    *,
    masking: MaskingOptions | None = None,
    error_display: ErrorDisplayOptions | None = None,
    loading: LoadingOptions | None = None,
    vault: VaultOptions | None = None,
    type_loaders: TypeLoaderMap | None = None,
) -> None:

Set global configuration. Pass dicts to override specific options: masking={"mask": "***"}, loading={"debug": True}. None parameters keep their current values. Empty dict {} resets the group to defaults.

Global config is also loaded from DATURE_* environment variables on first access.

MaskingConfig#

@dataclass(frozen=True, slots=True)
class MaskingConfig:
    mask: str = "<REDACTED>"
    visible_prefix: int = 0
    visible_suffix: int = 0
    min_heuristic_length: int = 8
    heuristic_threshold: float = 0.5
    secret_field_names: tuple[str, ...] = (
        "password",
        "passwd",
        "secret",
        "token",
        "api_key",
        "apikey",
        "api_secret",
        "access_key",
        "private_key",
        "auth",
        "credential",
    )
    mask_secrets: bool = True

Frozen dataclass controlling secret masking behavior.

ErrorDisplayConfig#

@dataclass(frozen=True, slots=True)
class ErrorDisplayConfig:
    max_visible_lines: int = 3
    max_line_length: int = 80

Frozen dataclass controlling error message formatting.

LoadingConfig#

@dataclass(frozen=True, slots=True)
class LoadingConfig:
    cache: bool | timedelta = True
    cache_engine: bool = False
    debug: bool = False
    nested_resolve_strategy: NestedResolveStrategy = "flat"
    expand_env_vars: ExpandEnvVarsMode = "default"
    search_system_paths: bool = True
    system_config_dirs: SystemConfigDirsArg = field(default_factory=_default_system_config_dirs)
    encoding: str | None = None

Frozen dataclass controlling load behavior defaults.

VaultConfig#

@dataclass(frozen=True, slots=True)
class VaultConfig:
    url: str | None = None
    token: str | None = None
    role_id: str | None = None
    secret_id: str | None = None
    namespace: str | None = None
    verify: bool | str = True
    mount_point: str = "secret"
    kv_version: Literal[1, 2] = 2

Frozen dataclass with connection defaults for VaultSource. Fields left unset on a VaultSource instance fall back to these values.

Validators#

Validators are built with the V DSL (dature.validators.v.V), used inside Annotated[T, <predicate>] field metadata:

from dataclasses import dataclass
from typing import Annotated

from dature import V


@dataclass
class ServiceConfig:
    port: Annotated[int, (V >= 1) & (V <= 65535)]
    name: Annotated[str, (V.len() >= 3) & (V.len() <= 50)]
    tags: Annotated[list[str], V.unique_items() & V.each(V.len() >= 3)]

All predicates are frozen dataclasses (frozen=True, slots=True) with two methods:

All predicates accept an optional error_message keyword to override the default message (placeholders like {value} / {pattern} are filled in from the predicate's own parameters). with_error_message(message) returns a copy with the message replaced — raises TypeError on composite predicates (&/\|/~), since they derive their message from their children.

Comparison#

V <op> value builds a ComparePredicate; V.len() <op> value builds a LengthComparePredicate (raises ValidatorTypeError at schema-build time if the field type doesn't support len()).

Other predicates#

Composition#

Predicates compose with & (AndPredicate), \| (OrPredicate), and ~ (NotPredicate) — all fail-fast on check_type/short-circuit on the validator function.

Root Validator#

"""Root validator: ``V.root(func)`` — cross-field checks.

``RootPredicate`` is intentionally **not** a :class:`Predicate`. Placing it in
``Annotated[...]`` metadata raises a ``TypeError`` at retort-build time — it may
only appear in ``Source.root_validators``.
"""

from collections.abc import Callable
from dataclasses import dataclass
from typing import final


@final
@dataclass(frozen=True, slots=True)
class RootPredicate:
    func: Callable[..., bool]
    error_message: str = "Root validation failed"

    def get_validator_func(self) -> Callable[..., bool]:
        return self.func

    def get_error_message(self) -> str:
        return self.error_message

V.root(func, error_message=...) builds a RootPredicate. Unlike the other predicates, RootPredicate is not a Predicate subclass and cannot be placed in Annotated[...] metadata — doing so raises TypeError at retort-build time. It may only be used in Source.root_validators or load(root_validators=...), and receives the fully-loaded dataclass instance:

Methods: get_validator_func() → returns func. get_error_message() → returns error_message.

Special Types#

SecretStr#

Module: dature.fields.secret_str. A string wrapper that hides its value in str() and repr().

ByteSize#

Module: dature.fields.byte_size. Parses human-readable byte sizes ("1.5 GB", "512 KiB") into an integer byte count.

Accepted formats: <number><unit> where unit is one of: B, KB, MB, GB, TB, PB (decimal) or KiB, MiB, GiB, TiB, PiB (binary). Case-insensitive. Whitespace between number and unit is allowed.

PaymentCardNumber#

Module: dature.fields.payment_card. Luhn-validated payment card number with brand detection.

Constructor strips spaces and dashes, validates digit-only 12–19 chars, and runs Luhn check. Raises ValueError on invalid input.

Other Type Aliases#

Source Classes#

File-based sources (inherit FileSource)#

All file-based sources accept the file parameter from FileSource plus all common parameters from Source.

file accepts str, Path, or file-like objects (StringIO, BytesIO, any TextIOBase/BufferedIOBase/RawIOBase). When file=None, the path defaults to the current directory.

file_display() returns the path as string, "<stream>" for file-like objects, or None when file=None.

JsonSource(FileSource)#

Json5Source(FileSource)#

Yaml11Source(FileSource)#

Yaml12Source(FileSource)#

Toml10Source(FileSource)#

Toml11Source(FileSource)#

IniSource(FileSource)#

Section headers become top-level dict keys. Dotted sections (database.pool) create nested dicts. prefix selects a single section.

Flat key-value sources (inherit FlatKeySource)#

All flat key-value sources accept nested_sep, nested_resolve_strategy and nested_resolve from FlatKeySource plus all common parameters from Source.

All values are strings. Automatic parsing of str, float, date, datetime, time, bytearray, bool, None, str | None. Nested JSON in values ([...], {...}) is inferred.

Nesting is built from nested_sep (default "__"): APP__DB__HOST=x → {"db": {"host": "x"}}.

EnvSource(FlatKeySource)#

Keys are lowercased after stripping prefix. resolve_location() returns env_var_name instead of file/line info.

EnvFileSource(FlatKeySource)#

Inherits from both FileFieldMixin and EnvSource, so accepts the file parameter. resolve_location() returns line range within the .env file.

DockerSecretsSource(FlatKeySource)#

Each file in dir_ becomes a key (filename, lowercased) with the file content (stripped) as value. Subdirectories are skipped. resolve_location() returns the path dir_/secret_name as file_path.

CLI sources (inherit CliSource)#

All CLI sources accept the common parameters from Source. CliSource overrides FlatKeySource's defaults: nested_sep defaults to "--" and expand_env_vars defaults to "disabled" (the shell already expands $VAR before argv reaches the parser).

ArgparseSource(CliSource)#

Unset defaults (argparse.SUPPRESS) are suppressed so they don't leak into merge mode; explicit bool flags are kept. Subcommands emit a discriminator field plus their own prefixed sub-arguments.

Remote sources (inherit RemoteSource)#

All remote sources accept the common parameters from Source. Values are fetched over the network via _fetch() rather than read from a local file.

VaultSource(RemoteSource)#

Requires either token or role_id+secret_id (raises at construction if neither or both are given). Raises KeyError on an invalid path, PermissionError on a forbidden/unauthorized response. See VaultConfig for global defaults.

Exceptions#

All exceptions are in dature.errors.

DatureError#

Base exception for all dature errors.

ValidatorTypeError(DatureError)#

Raised at schema-build time (not data-loading time) when a V predicate is incompatible with a field's type — e.g. V.len() applied to an int field. Signals that the schema itself is ill-formed; raised before any configuration data is read.

DatureErrorGroup(ExceptionGroup[DatureError])#

Base for all dature exception groups. Subclasses add domain-specific context; see DatureConfigError, EnvVarExpandError, CrossRefExpandError below.

DatureConfigError(DatureErrorGroup)#

Aggregated config loading errors. Contains one or more FieldLoadError sub-exceptions.

str() returns "<name> loading errors (<count>)".

FieldLoadError(DatureError)#

Single field loading error with source location.

str() returns a formatted multi-line message with source context and caret pointing at the value.

SourceLoadError(DatureError)#

Source-level loading failure (e.g. file not found, parse error).

MergeConflictError(DatureConfigError)#

Raised with strategy="raise_on_conflict" when sources provide different values. Contains MergeConflictFieldError sub-exceptions.

MergeConflictFieldError(DatureError)#

Per-field merge conflict.

FieldGroupError(DatureConfigError)#

Field group constraint violation. Contains FieldGroupViolationError sub-exceptions.

FieldGroupViolationError(DatureError)#

Single field group violation.

EnvVarExpandError(DatureErrorGroup)#

Missing environment variables in expand_env_vars="strict" mode. Contains MissingEnvVarError sub-exceptions. The exception actually raised during config loading is ConfigEnvVarExpandError (below), which also carries dataclass_name.

ConfigEnvVarExpandError(EnvVarExpandError, DatureConfigError)#

The concrete exception raised by load()/Loader.load() when env var expansion fails in strict mode. str() returns "<dataclass_name> env expand errors (<count>)".

MissingEnvVarError(DatureError)#

Single missing env var.

CrossRefError(DatureError)#

Single failure resolving a ${@tag.key} cross-source reference — see Cross-Source References.

CrossRefExpandError(DatureErrorGroup)#

Aggregated cross-source reference errors. Contains one or more CrossRefError sub-exceptions.

SourceLocation#

Frozen dataclass used in error messages to point at the source of a value.

LineRange#

Frozen dataclass for file line ranges.

repr() returns "line 5" or "line 5-8".

CaretSpan#

Frozen dataclass describing a ^^^ caret span within a SourceLocation.line_content line. Re-exported via dature.errors.

Type Aliases#