Merge two configuration trees with per-node rules

01The design, in the six-step framework

1. Requirements and use cases

A platform has a base configuration and an overlay for each environment, both parsed from YAML into nested dicts and lists. A developer merges the two trees. By default maps merge recursively and the overlay wins on scalars. Certain nodes need their own rule: lists of ports are unioned, containers are matched by name, a security setting must fail if the two sides disagree. New rules must be addable without touching the merge walk.

Out of scope: reading and writing YAML text, anchors and comments, and merging more than two trees, which is a fold over this operation.

Constraints: inputs are never mutated, because a base config is shared across environments; the result must not alias either input; the outcome is deterministic, including which rule applies when two patterns match.

2. Core entities and relationships

• Node: the data, plain dicts, lists and scalars. There is no wrapper class, because a parsed YAML tree is already a composite.
• MergeRule: one way to combine two values at a path.
• DeepMap, OverlayWins, BaseWins, Fail, Append, UnionList, MergeByKey: the built-in rules.
• Merger: the registry and dispatcher. It picks the rule for a node and offers recurse so that container rules can descend.
• MergeConflict: names the path where two values could not be reconciled.

The merger aggregates rule objects registered by the caller. The dependency runs both ways by design: Merger calls a rule, and container rules call back into ctx.recurse, which is how rule selection applies at every depth while the walk stays in one place.

3. Interfaces and key methods

class MergeRule(Protocol):
    def merge(self, base: Node, overlay: Node, path: Path, ctx: "Merger") -> Node:
        """Return the merged value for one node; must not mutate its inputs."""

class Merger:
    def __init__(self, default: MergeRule, by_type: Mapping[type, MergeRule] = ...) -> None: ...
    def at(self, pattern: str, rule: MergeRule) -> "Merger":
        """Register a rule for a path pattern such as 'services.*.ports'."""
    def merge(self, base: Node, overlay: Node) -> Node:
        """Deep-merge two trees; the most specific rule wins."""
    def recurse(self, base: Node, overlay: Node, path: Path) -> Node: ...

Invariants: merge does not change base or overlay; every value in the result is a copy; a key present on one side only is carried over unchanged; rule choice depends only on the path and the value types, never on registration order for patterns of different specificity.

4. Design patterns and the choices behind them

A strategy per node

What varies is how two values combine, and it varies by location. A recursive function with a growing if-chain on key names was rejected, because each new setting would edit the core walk. Each behaviour is a strategy object, and most are three lines.

A registry with a resolution order

The reported problem expected an open-ended rule set, so rule selection is a registry lookup with three tiers: a path pattern if one matches, then a rule for the value's type, then the default. When the two sides have different types, for example a list replaced by a scalar, the default applies and no recursion happens. Among matching patterns, the one with fewest wildcards and then the longest wins, so services.api.ports beats services.*.ports wherever it was registered. Adding a rule never edits Merger, which is the open-closed test.

Rules receive the merger as context

MergeByKey has to merge two matched list items, and those items may contain nested rules of their own. Passing ctx lets a rule hand children back to the dispatcher, with the path extended. The alternative, letting each container rule walk its own subtree, would skip path rules beneath it.

Deep copies at the boundary

Returning a reference from a rule would alias the input, and a later edit to the result would change the shared base. Each leaf rule copies. This costs time on large trees, and an interviewer may accept shared immutable values instead; state the choice.

5. Core flow and code

1. The caller builds Merger().at("service.ports", UnionList()).at("service.containers", MergeByKey("name")).at("security.*", Fail()).
2. merge(base, overlay) calls recurse with an empty path. No pattern matches, both sides are dicts, so DeepMap runs.
3. DeepMap copies keys found on one side and calls ctx.recurse for shared keys with the path extended.
4. At service.ports, the pattern selects UnionList, which keeps base order and appends new entries.
5. At service.containers, MergeByKey matches items by name and recurses into each matched pair, so service.containers.web.image follows the scalar default.
6. At security.tls, Fail compares the values; a disagreement raises MergeConflict naming the path.

from __future__ import annotations
import copy
from fnmatch import fnmatchcase
from typing import Mapping, Protocol, Union

Node = Union[dict, list, str, int, float, bool, None]
Path = tuple[str, ...]

class MergeConflict(Exception):
    def __init__(self, path: Path, base: Node, overlay: Node) -> None:
        super().__init__(f"conflict at {'.'.join(path)}: {base!r} vs {overlay!r}")
        self.path = path

class MergeRule(Protocol):
    def merge(self, base: Node, overlay: Node, path: Path, ctx: "Merger") -> Node: ...

class OverlayWins:
    def merge(self, base, overlay, path, ctx):
        return copy.deepcopy(overlay)

class BaseWins:
    def merge(self, base, overlay, path, ctx):
        return copy.deepcopy(base)

class Fail:
    def merge(self, base, overlay, path, ctx):
        if base != overlay:
            raise MergeConflict(path, base, overlay)
        return copy.deepcopy(base)

class Append:
    def merge(self, base, overlay, path, ctx):
        return copy.deepcopy(base) + copy.deepcopy(overlay)

class UnionList:
    """Keep order, drop repeats; items may be unhashable, so compare by equality."""
    def merge(self, base, overlay, path, ctx):
        out = copy.deepcopy(base)
        out.extend(copy.deepcopy(x) for x in overlay if x not in out)
        return out

class MergeByKey:
    """Lists of maps matched on one key: containers, ports, routes."""
    def __init__(self, key: str) -> None:
        self.key = key
    def merge(self, base, overlay, path, ctx):
        out = copy.deepcopy(base)
        index = {item[self.key]: i for i, item in enumerate(out)}
        for item in overlay:
            k = item[self.key]
            if k in index:
                out[index[k]] = ctx.recurse(out[index[k]], item, path + (str(k),))
            else:
                index[k] = len(out)
                out.append(copy.deepcopy(item))
        return out

class DeepMap:
    def merge(self, base, overlay, path, ctx):
        out = {}
        for key in base.keys() | overlay.keys():
            if key not in overlay:
                out[key] = copy.deepcopy(base[key])
            elif key not in base:
                out[key] = copy.deepcopy(overlay[key])
            else:
                out[key] = ctx.recurse(base[key], overlay[key], path + (key,))
        return dict(sorted(out.items(), key=lambda kv: str(kv[0])))

class Merger:
    def __init__(self, default: MergeRule | None = None,
                 by_type: Mapping[type, MergeRule] | None = None) -> None:
        self._default = default or OverlayWins()
        self._by_type = {dict: DeepMap(), list: OverlayWins(), **(by_type or {})}
        self._paths: list[tuple[str, MergeRule]] = []

    def at(self, pattern: str, rule: MergeRule) -> "Merger":
        self._paths.append((pattern, rule))
        return self

    def _rule_for(self, base: Node, overlay: Node, path: Path) -> MergeRule:
        dotted = ".".join(path)
        hits = [(p, r) for p, r in self._paths if fnmatchcase(dotted, p)]
        if hits:                                      # most specific: fewest wildcards, then longest
            return min(hits, key=lambda h: (h[0].count("*"), -len(h[0])))[1]
        if overlay is None or type(base) is not type(overlay):
            return self._default                      # type change: never recurse
        return self._by_type.get(type(base), self._default)

    def recurse(self, base: Node, overlay: Node, path: Path) -> Node:
        return self._rule_for(base, overlay, path).merge(base, overlay, path, self)

    def merge(self, base: Node, overlay: Node) -> Node:
        return self.recurse(base, overlay, ())

6. Extensions, concurrency and what interviewers probe

• Delete markers: overlays often need to remove a key. Add a sentinel value and handle it in DeepMap; plain null is a bad marker, because null may be a legitimate value, and that distinction must be agreed up front.
• Merge provenance: record which file each leaf came from by threading a recorder through ctx; useful for debugging production config.
• Rules from YAML: a name-to-rule table lets the config declare its own merge rules; only a fixed registry should be reachable, not arbitrary classes.
• Concurrency: a Merger is read-only after setup and merge touches only its inputs, so one instance can serve many threads if registration finishes first.
• Test first: scalar override, key on one side, nested maps, type mismatch, pattern precedence, MergeByKey with a nested rule, and an assertion that the inputs are unchanged after merging.
• Common mistake: mutating the base dict in place, which works in a demo and corrupts the second environment.

02Where this question was reported

• PracHub · Tagged Low-Level Design (OOP); per-node strategy