Merging and typed results

Merge strategies, typed hydration and dot access, all importable from anycfg. See Merging and Typed results.

Merging

anycfg.DELETE = _Delete() module-attribute

anycfg.INHERIT = _Inherit() module-attribute

anycfg.Merge

Bases: Protocol

A merge strategy: combine a with the later document b.

__call__(a, b, *, memo=None, **options)

Source code in src/anycfg/_merge.py

def __call__(
    self,
    a: _ty.Any,
    b: _ty.Any,
    *,
    memo: "_ty.Optional[dict]" = None,
    **options: _ty.Any,
) -> _ty.Any: ...

anycfg.register_merge(name, strategy, *, replace=False)

Make strategy selectable as merge=name.

Raises ValueError when name is taken, unless replace is true.

Source code in src/anycfg/_merge.py

def register_merge(name: str, strategy: Merge, *, replace: bool = False) -> None:
    """Make *strategy* selectable as ``merge=name``.

    Raises `ValueError` when *name* is taken, unless *replace* is true.
    """
    global _STRATEGIES
    key = name.lower()
    with _lock:
        if key in _STRATEGIES and not replace:
            raise ValueError(
                f"merge strategy {key!r} is already registered; pass replace=True"
            )
        # Replaced whole, so a reader never sees the table change under it.
        _STRATEGIES = {**_STRATEGIES, key: strategy}

anycfg.merge_strategies()

Every registered strategy name, sorted.

Source code in src/anycfg/_merge.py

def merge_strategies() -> "_ty.List[str]":
    """Every registered strategy name, sorted."""
    return sorted(_STRATEGIES)

Typed results

anycfg.hydrate(cls, data, *, extra='ignore')

Build one cls from data, coercing every field through its annotation.

The single-layer form of :func:typed_merge, and what load(..., into=cls) uses: a dataclass, a namespace, a mapping type, a TypedDict, a Pydantic model (validated by Pydantic) or any class whose constructor takes the fields as keywords.

Raises:

Type	Description
ConfigTypeError	When data cannot be built into cls, for an unknown key under extra="forbid", or a TypedDict's missing required key.
ConfigValueError	When a value cannot be coerced, or data contains itself.

Source code in src/anycfg/_hydrate.py

def hydrate(cls: "typing.Type[T]", data: object, *, extra: str = "ignore") -> T:
    """Build one *cls* from *data*, coercing every field through its annotation.

    The single-layer form of :func:`typed_merge`, and what
    ``load(..., into=cls)`` uses: a dataclass, a namespace, a mapping type, a
    TypedDict, a Pydantic model (validated by Pydantic) or any class whose
    constructor takes the fields as keywords.

    Raises:
        ConfigTypeError: When *data* cannot be built into *cls*, for an
            unknown key under ``extra="forbid"``, or a TypedDict's missing
            required key.
        ConfigValueError: When a value cannot be coerced, or *data* contains
            itself.
    """
    if data is None:
        raise ConfigTypeError(
            f"cannot build {getattr(cls, '__name__', cls)} from no data"
        )
    return typed_merge(cls, data, extra=extra)

anycfg.dehydrate(obj, memo=None)

obj as plain data a format can write: the inverse of :func:hydrate.

A type with a _dump_() method gives its own plain form; a dataclass, a namespace or another object with fields becomes a mapping; an Enum member its value; a path, a Decimal or a UUID its text; mappings, lists and tuples are walked. Anything else is returned as it is, for the format to accept or refuse. Aliases and cycles are kept.

Source code in src/anycfg/_hydrate.py

def dehydrate(obj: object, memo: "typing.Optional[dict]" = None) -> object:
    """*obj* as plain data a format can write: the inverse of :func:`hydrate`.

    A type with a ``_dump_()`` method gives its own plain form; a dataclass, a
    namespace or another object with fields becomes a mapping; an ``Enum``
    member its value; a path, a `Decimal` or a `UUID` its text; mappings, lists
    and tuples are walked.
    Anything else is returned as it is, for the format to accept or refuse.
    Aliases and cycles are kept.
    """
    if isinstance(obj, ScalarText):
        return obj.value
    if obj is None or type(obj) in (str, int, float, bool):
        return obj
    memo = {} if memo is None else memo
    hit = memo.get(id(obj))
    if hit is not None:
        return hit[1]
    dump_hook = getattr(type(obj), "_dump_", None)
    if callable(dump_hook):
        return dehydrate(obj._dump_(), memo)
    if isinstance(obj, enum.Enum):
        return dehydrate(obj.value, memo)
    if isinstance(obj, (pathlib.PurePath, decimal.Decimal, uuid.UUID)):
        # Text no format misreads, and `hydrate` builds the type from it again.
        return str(obj)
    if isinstance(obj, _abc.Mapping):
        result: dict = {}
        memo[id(obj)] = (obj, result)
        for key, value in obj.items():
            result[dehydrate(key, memo)] = dehydrate(value, memo)
        return result
    if isinstance(obj, (list, tuple)) and not hasattr(obj, "_fields"):
        items: list = []
        memo[id(obj)] = (obj, items)
        items.extend(dehydrate(item, memo) for item in obj)
        return items
    if dataclasses.is_dataclass(obj) and not isinstance(obj, type):
        fields = {f.name: getattr(obj, f.name) for f in dataclasses.fields(obj)}
        return dehydrate(fields, memo)
    if isinstance(obj, tuple):  # a NamedTuple
        return dehydrate(obj._asdict(), memo)
    if isinstance(obj, Namespace):
        return dehydrate(vars(obj), memo)
    return obj

anycfg.typed_merge(cls, *objects, init=True, extra='ignore')

typed_merge(
    cls: "typing.Type[T]",
    *,
    init: bool = True,
    extra: str = "ignore"
) -> None

typed_merge(
    cls: "typing.Type[T]",
    *objects: object,
    init: bool = True,
    extra: str = "ignore"
) -> T

typed_merge(
    cls: typing.Any,
    *objects: object,
    init: bool = True,
    extra: str = "ignore"
) -> typing.Any

Merge objects, in order, into one instance of cls.

Type-guided: a mapping, dataclass or namespace merges field by field, each field through its own annotation; a sequence or set takes the last layer's value with every item coerced; a scalar takes the last value, coerced through the hint. None layers are skipped at every level, so a later None never overrides an earlier value; all-None (or no) layers give None. A layer of another shape replaces everything before it. A node reached twice is built once; one that contains itself raises ConfigValueError.

Coercion never loses data: int accepts a whole-number float or an integer string, never 1.9 or a bool; bool reads true/yes/on/1 and false/no/off/0; date/datetime/time read ISO strings; an Enum takes a value, then a member name; a Literal takes one of its values; a NewType coerces as its base type. A NamedTuple accepts a sequence or a mapping of its fields. A Pydantic model's layers are deep-merged and validated by Pydantic.

extra decides what a key the target class (or TypedDict) does not accept does: "ignore" drops it, "forbid" raises ConfigTypeError naming it. A constructor taking **kwargs accepts every key. init false builds the instance without calling __init__.

Hooks: a classmethod __merge__(cls, *objects, init=True) takes over a type's merging entirely (see OpaqueMerge); a classmethod _load_(cls, value) builds the type from one value (a scalar as written in the document when that is known, otherwise plain data); a _parse_<field>(value) method coerces that field as it is collected from each layer. Without a hook the basic coercions above apply.

Source code in src/anycfg/_hydrate.py

def typed_merge(
    cls: typing.Any, *objects: object, init: bool = True, extra: str = "ignore"
) -> typing.Any:
    """Merge *objects*, in order, into one instance of *cls*.

    Type-guided: a mapping, dataclass or namespace merges field by field, each
    field through its own annotation; a sequence or set takes the last layer's
    value with every item coerced; a scalar takes the last value, coerced
    through the hint. ``None`` layers are skipped at every level, so a later
    ``None`` never overrides an earlier value; all-``None`` (or no) layers give
    ``None``. A layer of another shape replaces everything before it. A node
    reached twice is built once; one that contains itself raises
    `ConfigValueError`.

    Coercion never loses data: ``int`` accepts a whole-number float or an
    integer string, never ``1.9`` or a bool; ``bool`` reads ``true/yes/on/1``
    and ``false/no/off/0``; ``date``/``datetime``/``time`` read ISO strings;
    an ``Enum`` takes a value, then a member name; a ``Literal`` takes one of
    its values; a ``NewType`` coerces as its base type. A ``NamedTuple``
    accepts a sequence or a mapping of its fields. A Pydantic model's layers
    are deep-merged and validated by Pydantic.

    *extra* decides what a key the target class (or TypedDict) does not accept
    does: ``"ignore"`` drops it, ``"forbid"`` raises `ConfigTypeError` naming
    it. A constructor taking ``**kwargs`` accepts every key. *init* false
    builds the instance without calling ``__init__``.

    Hooks: a classmethod ``__merge__(cls, *objects, init=True)`` takes over a
    type's merging entirely (see `OpaqueMerge`); a classmethod
    ``_load_(cls, value)`` builds the type from one value (a scalar as written
    in the document when that is known, otherwise plain data); a
    ``_parse_<field>(value)`` method coerces that field as it is collected
    from each layer. Without a hook the basic coercions above apply.
    """
    if extra not in _EXTRA_MODES:
        raise ValueError(f"extra= takes {' or '.join(_EXTRA_MODES)}, not {extra!r}")
    return _Run(init, extra).run(cls, objects)

anycfg.OpaqueMerge

Mixin marking a type opaque to :func:typed_merge: the last layer wins.

For a fully built object whose __init__ already normalized its fields, and which should not be re-merged field by field — including under Optional[...] and other unions.

__merge__(*objects, init=True) classmethod

Source code in src/anycfg/_hydrate.py

@classmethod
def __merge__(cls, *objects: object, init: bool = True) -> object:
    return _last_wins(cls, *objects, init=init)

anycfg.opaque(cls)

Class decorator equivalent of :class:OpaqueMerge, leaving the bases alone.

Source code in src/anycfg/_hydrate.py

def opaque(cls: "typing.Type[T]") -> "typing.Type[T]":
    """Class decorator equivalent of :class:`OpaqueMerge`, leaving the bases alone."""
    cls.__merge__ = classmethod(_last_wins)
    return cls

anycfg.TypedNamespace(**kwargs)

Bases: Namespace

An :class:argparse.Namespace that applies _parse_<field> coercers.

At construction each keyword's value goes through a _parse_<name> method when the class defines one, so a built instance is already normalized. :func:typed_merge applies the same hooks to raw layers, skips them for a layer that is already a TypedNamespace, and assembles the result without calling __init__ — so every value is parsed exactly once and a parser need not be idempotent.

Source code in src/anycfg/_hydrate.py

def __init__(self, **kwargs: object) -> None:
    for name, value in list(kwargs.items()):
        parser = getattr(self, f"_parse_{name}", None)
        if parser is not None:
            kwargs[name] = parser(value)
    super().__init__(**kwargs)

Dot access

anycfg.DotAccessibleDict(*args, **kwargs)

Bases: dict

Dictionary subclass supporting dot-notation queries and attribute access.

Nested mappings are converted once, at construction, so item access, attribute access and object identity all agree however the value is reached, whatever the read order. Reads never write: get() follows dict.get's contract and a miss returns the default without storing it.

Values assigned after construction are stored exactly as given — a plain dict written with cfg["x"] = {...} stays a plain dict.

A key that collides with something the class defines (items, get, copy, ...) is reachable only through item access: attribute reads find the method, so attribute writes and deletes of such a name raise AttributeError rather than letting the two disagree. A subclass that needs a real instance attribute uses object.__setattr__.

dict(cfg) is the way to get a plain dict back.

Source code in src/anycfg/_dotaccess.py

def __init__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
    super().__init__(*args, **kwargs)
    # Seed the memo with this object, and with the source mapping when
    # there is one, so a self-referencing source resolves to this instance.
    memo = {id(self): self}
    if len(args) == 1 and isinstance(args[0], dict):
        memo[id(args[0])] = self
    for key, value in list(self.items()):
        dict.__setitem__(self, key, _to_dot_access(value, memo))

__delattr__(name)

Source code in src/anycfg/_dotaccess.py

def __delattr__(self, name: str) -> None:
    if hasattr(type(self), name):
        raise AttributeError(
            f"{name!r} is read-only on {type(self).__name__!r} because the class "
            f"defines it; use del cfg[{name!r}] to remove that key"
        )
    try:
        del self[name]
    except KeyError:
        raise AttributeError(
            f"{type(self).__name__!r} object has no attribute {name!r}"
        ) from None

__getattr__(name)

Source code in src/anycfg/_dotaccess.py

def __getattr__(self, name: str) -> typing.Any:
    try:
        return self[name]
    except KeyError:
        raise AttributeError(
            f"{type(self).__name__!r} object has no attribute {name!r}"
        ) from None

__or__(other)

Source code in src/anycfg/_dotaccess.py

def __or__(self, other: typing.Any) -> typing.Any:
    if not isinstance(other, dict):
        return NotImplemented
    merged = type(self)()
    # Built like dict's own operator: key order of the left operand, then
    # the right's, right wins. Values are stored as given -- conversion
    # happens at construction only.
    for mapping in (self, other):
        for key, value in mapping.items():
            dict.__setitem__(merged, key, value)
    return merged

__ror__(other)

Source code in src/anycfg/_dotaccess.py

def __ror__(self, other: typing.Any) -> typing.Any:
    if not isinstance(other, dict):
        return NotImplemented
    merged = type(self)()
    for mapping in (other, self):
        for key, value in mapping.items():
            dict.__setitem__(merged, key, value)
    return merged

__setattr__(name, value)

Source code in src/anycfg/_dotaccess.py

def __setattr__(self, name: str, value: object) -> None:
    # A dict subclass cannot expose "items", "keys", "get" and friends as
    # attributes, so refusing the write is the only way attribute reads and
    # writes cannot disagree about such a name. Use cfg[name] for the key;
    # object.__setattr__ for a real instance attribute in a subclass.
    if hasattr(type(self), name):
        raise AttributeError(
            f"{name!r} is read-only on {type(self).__name__!r} because the class "
            f"defines it; use cfg[{name!r}] to set that key"
        )
    self[name] = value

copy()

A shallow copy of the same class — like dict.copy, but typed.

Source code in src/anycfg/_dotaccess.py

def copy(self) -> "DotAccessibleDict":
    """A shallow copy of the same class — like `dict.copy`, but typed."""
    return _copy.copy(self)

dig(*keys, default=None)

:func:anycfg.dig on this mapping: keys as given, a miss is default.

Source code in src/anycfg/_dotaccess.py

def dig(self, *keys: typing.Any, default: typing.Any = None) -> typing.Any:
    """:func:`anycfg.dig` on this mapping: keys as given, a miss is *default*."""
    return dig(self, *keys, default=default)

get(key, default=None, dig=True)

Look up key, optionally as a path into nested containers.

The order is: an exact key wins; then, when dig is true, a dotted-string or tuple path is traversed; otherwise default.

A dotted string ("database.credentials.user") walks mappings by segment. A segment of ASCII digits indexes a list or tuple ("servers.0.host"), but on a mapping it is always the string key, so "codes.0" never means the integer 0. Exact-key precedence applies at the top level only, so a nested key that itself contains a dot is unreachable this way — pass a tuple path for that (("metadata", "labels", "app.kubernetes.io/name")), which also reaches integer keys and uses ints, not digit strings, as indexes.

A None (or any non-traversable value) reached with segments still to go yields default; the final segment's value is returned as stored, None included — so an explicit null stays distinguishable from an absent key.

Never inserts: a miss returns default itself, as dict.get does.

Source code in src/anycfg/_dotaccess.py

def get(
    self, key: typing.Hashable, default: typing.Any = None, dig: bool = True
) -> typing.Any:
    """Look up *key*, optionally as a path into nested containers.

    The order is: an **exact** key wins; then, when *dig* is true, a
    dotted-string or tuple path is traversed; otherwise *default*.

    A dotted string (``"database.credentials.user"``) walks mappings by
    segment. A segment of ASCII digits indexes a list or tuple
    (``"servers.0.host"``), but on a mapping it is always the *string* key,
    so ``"codes.0"`` never means the integer ``0``. Exact-key precedence
    applies at the top level only, so a nested key that itself contains a
    dot is unreachable this way — pass a **tuple** path for that
    (``("metadata", "labels", "app.kubernetes.io/name")``), which also
    reaches integer keys and uses ints, not digit strings, as indexes.

    A ``None`` (or any non-traversable value) reached with segments still
    to go yields *default*; the final segment's value is returned as
    stored, ``None`` included — so an explicit ``null`` stays
    distinguishable from an absent key.

    Never inserts: a miss returns *default* itself, as `dict.get` does.
    """
    if key in self:
        return super().get(key, default)
    if not dig:
        return default

    if isinstance(key, str):
        if "." not in key:
            return default
        # Inlined rather than delegated: this is the benchmarked read path,
        # and a helper call with keyword arguments showed up in it.
        segments = key.split(".")
        current: object = self
        last = len(segments) - 1
        for index, segment in enumerate(segments):
            if isinstance(current, dict):
                # dict.get, so a defaultdict stored later does not grow a
                # key through a read, and __missing__ is never consulted.
                current = dict.get(current, segment, _MISSING)
                if current is _MISSING:
                    return default
            elif current.__class__ is list or current.__class__ is tuple:
                # ASCII digits only: a Unicode digit must not reach int(),
                # and a negative or non-numeric segment ends the path.
                if not (segment.isascii() and segment.isdigit()):
                    return default
                position = int(segment)
                if position >= len(current):
                    return default
                current = current[position]
            else:
                # A scalar, a None, or a non-dict Mapping: the path stops.
                return default
            if current is None and index != last:
                return default
        return current
    if isinstance(key, tuple):
        if not key:
            return default
        return self._dig(key, default)
    # Any other hashable key: absent is absent. (dict.get's contract; an
    # unhashable one has already raised TypeError above.)
    return default