Merge Strategies#

Per-Field Merge Strategies#

Override the global strategy for individual fields using field_merges. Each value can be one of the built-in strategy names below, or any callable or custom class implementing FieldMergeStrategy.

Available field merge strategies:

Strategy	Behavior
`"first_wins"`	Keep the value from the first source
`"last_wins"`	Keep the value from the last source
`"append"`	Concatenate lists: `base + override`
`"append_unique"`	Concatenate lists, removing duplicates
`"prepend"`	Concatenate lists: `override + base`
`"prepend_unique"`	Concatenate lists in reverse order, removing duplicates

Given two sources with overlapping tags:

merging_field_base.yamlmerging_field_override.yaml

tags:
  - "web"
  - "default"

tags:
  - "web"
  - "api"

Each strategy produces a different result:

first_winslast_winsappendappend_uniqueprependprepend_unique

from dataclasses import dataclass

import dature


@dataclass
class Config:
    tags: list[str]


config = dature.load(
    dature.Yaml12Source(file=SOURCES_DIR / "merging_field_base.yaml"),
    dature.Yaml12Source(file=SOURCES_DIR / "merging_field_override.yaml"),
    schema=Config,
    # UNIQUE(a + b, keep="first")
    field_merges={dature.F[Config].tags: "first_wins"},
)

assert config.tags == ["web", "default"]
assert config.tags == ["web", "default"]

from dataclasses import dataclass

import dature


@dataclass
class Config:
    tags: list[str]


config = dature.load(
    dature.Yaml12Source(file=SOURCES_DIR / "merging_field_base.yaml"),
    dature.Yaml12Source(file=SOURCES_DIR / "merging_field_override.yaml"),
    schema=Config,
    # UNIQUE(a + b, keep="last")
    field_merges={dature.F[Config].tags: "last_wins"},
)

assert config.tags == ["web", "api"]

from dataclasses import dataclass

import dature


@dataclass
class Config:
    tags: list[str]


config = dature.load(
    dature.Yaml12Source(file=SOURCES_DIR / "merging_field_base.yaml"),
    dature.Yaml12Source(file=SOURCES_DIR / "merging_field_override.yaml"),
    schema=Config,
    field_merges={dature.F[Config].tags: "append"},  # a + b
)

assert config.tags == ["web", "default", "web", "api"]

from dataclasses import dataclass

import dature


@dataclass
class Config:
    tags: list[str]


config = dature.load(
    dature.Yaml12Source(file=SOURCES_DIR / "merging_field_base.yaml"),
    dature.Yaml12Source(file=SOURCES_DIR / "merging_field_override.yaml"),
    schema=Config,
    field_merges={dature.F[Config].tags: "append_unique"},  # UNIQUE(a + b)
)

assert config.tags == ["web", "default", "api"]

from dataclasses import dataclass

import dature


@dataclass
class Config:
    tags: list[str]


config = dature.load(
    dature.Yaml12Source(file=SOURCES_DIR / "merging_field_base.yaml"),
    dature.Yaml12Source(file=SOURCES_DIR / "merging_field_override.yaml"),
    schema=Config,
    field_merges={dature.F[Config].tags: "prepend"},  # b + a
)

assert config.tags == ["web", "api", "web", "default"]

from dataclasses import dataclass

import dature


@dataclass
class Config:
    tags: list[str]


config = dature.load(
    dature.Yaml12Source(file=SOURCES_DIR / "merging_field_base.yaml"),
    dature.Yaml12Source(file=SOURCES_DIR / "merging_field_override.yaml"),
    schema=Config,
    field_merges={dature.F[Config].tags: "prepend_unique"},  # UNIQUE(b + a)
)

assert config.tags == ["web", "api", "default"]

Nested fields are supported — see Field Paths for the full syntax.

With `raise_on_conflict`#

Fields with an explicit strategy are excluded from conflict detection:

Pythoncommon_defaults.yamlcommon_overrides.yaml

from dataclasses import dataclass

import dature


@dataclass
class Config:
    host: str
    port: int
    tags: list[str]


config = dature.load(
    dature.Yaml12Source(file=SHARED_DIR / "common_defaults.yaml"),
    dature.Yaml12Source(file=SHARED_DIR / "common_overrides.yaml"),
    schema=Config,
    strategy="raise_on_conflict",
    field_merges={
        dature.F[Config].host: "last_wins",
        dature.F[Config].port: "last_wins",
        dature.F[Config].tags: "append_unique",
    },
)

assert config.host == "production.example.com"
assert config.port == 8080
assert config.tags == ["default", "web", "api"]

host: "localhost"
port: 3000
tags:
  - "default"

host: "production.example.com"
port: 8080
tags:
  - "web"
  - "api"

Custom Field Strategy#

The `FieldMergeStrategy` Protocol#

Any callable that takes a list[JSONValue] (one value per source) and returns the merged value satisfies the public FieldMergeStrategy Protocol:

@runtime_checkable
class FieldMergeStrategy(Protocol):
    def __call__(self, values: list[JSONValue]) -> JSONValue: ...

The built-in field strategies are also exposed as classes from dature.strategies.field: FieldFirstWins, FieldLastWins, FieldAppend, FieldAppendUnique, FieldPrepend, FieldPrependUnique. They satisfy the same Protocol, so you can pass them directly to field_merges or compose them inside your own strategy.

Examples#

Pick a plain function for one-off logic, or a class for a named, reusable reducer:

FunctionClasscommon_defaults.yamlcommon_overrides.yaml

from dataclasses import dataclass
from typing import Any

import dature


@dataclass
class Config:
    host: str
    port: int
    tags: list[str]


def merge_tags(values: list[Any]) -> list[str]:
    return sorted({v for lst in values for v in lst})


config = dature.load(
    dature.Yaml12Source(file=SHARED_DIR / "common_defaults.yaml"),
    dature.Yaml12Source(file=SHARED_DIR / "common_overrides.yaml"),
    schema=Config,
    strategy="last_wins",
    field_merges={dature.F[Config].tags: merge_tags},
)

assert config.host == "production.example.com"
assert config.port == 8080
assert config.tags == ["api", "default", "web"]

from dataclasses import dataclass
from typing import cast

import dature
from dature.strategies.field import FieldMergeStrategy
from dature.type_aliases import JSONValue


@dataclass
class Config:
    host: str
    port: int
    tags: list[str]


class SortedUnion:
    def __call__(self, values: list[JSONValue]) -> JSONValue:
        merged: set[str] = set()
        for chunk in values:
            if isinstance(chunk, list):
                merged.update(str(v) for v in chunk)
        return cast("JSONValue", sorted(merged))


strategy: FieldMergeStrategy = SortedUnion()

config = dature.load(
    dature.Yaml12Source(file=SHARED_DIR / "common_defaults.yaml"),
    dature.Yaml12Source(file=SHARED_DIR / "common_overrides.yaml"),
    schema=Config,
    field_merges={dature.F[Config].tags: strategy},
)

assert config.tags == ["api", "default", "web"]

host: "localhost"
port: 3000
tags:
  - "default"

host: "production.example.com"
port: 8080
tags:
  - "web"
  - "api"

Custom Source Strategy#

The global strategy parameter accepts not only the names from Merge Strategies but also any object implementing the public SourceMergeStrategy Protocol:

@runtime_checkable
class SourceMergeStrategy(Protocol):
    def __call__(self, sources: Sequence[SourceProtocol], ctx: LoadCtx) -> JSONValue: ...

The strategy receives the raw Source instances (not pre-loaded data) and a LoadCtx helper. The primary API for applying a source to the running base is ctx.merge(source=src, base=base, op=...) — it loads the source (cached), runs the merge op (default deep_merge_last_wins), and registers the step so debug logs and LoadReport.field_origins are populated correctly. A minimal custom strategy is one loop (as example of SourceLastWins):

class SourceLastWins:
    def __call__(self, sources: Sequence[SourceProtocol], ctx: LoadCtx) -> JSONValue:
        base: JSONValue = {}
        for idx in range(len(sources)):
            base = ctx.merge(source_idx=idx, base=base)
        return base

Override op to plug in your own merge function — e.g. shallow overlay for env on top of files:

Pythoncommon_defaults.yamlcommon_overrides.yaml

from collections.abc import Sequence
from dataclasses import dataclass

import dature
from dature.sources.protocol import SourceProtocol
from dature.strategies import LoadCtx, SourceMergeStrategy
from dature.type_aliases import JSONValue


@dataclass
class Config:
    host: str
    port: int
    tags: list[str]


def _dict_overlay(a: JSONValue, b: JSONValue) -> JSONValue:
    return {**a, **b} if isinstance(a, dict) and isinstance(b, dict) else b


class EnvOverrides:
    def __call__(
        self,
        sources: Sequence[SourceProtocol],
        ctx: LoadCtx,
    ) -> JSONValue:
        base: JSONValue = {}
        for idx, s in enumerate(sources):
            if isinstance(s, dature.EnvSource):
                base = ctx.merge(source_idx=idx, base=base, op=_dict_overlay)
            else:
                base = ctx.merge(source_idx=idx, base=base)
        return base


strategy: SourceMergeStrategy = EnvOverrides()

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

assert config.host == "production.example.com"
assert config.port == 8080

host: "localhost"
port: 3000
tags:
  - "default"

host: "production.example.com"
port: 8080
tags:
  - "web"
  - "api"

isinstance(src, EnvSource) (or any other concrete Source subclass) lets the strategy dispatch on source type — useful when env variables should override file content rather than merge with it. Pass skip_on_error=True to ctx.merge(...) (or ctx.load(...)) if you want broken sources to be skipped silently regardless of skip_if_broken (this is what SourceFirstFound does internally).

ctx.merge is the single hook — once your strategy funnels every per-source step through it, debug logs ([Cls] Merge step N ..., State after step N: ...) and LoadReport.field_origins are populated automatically; there's no separate registration call to remember.