Utilities

Shared building blocks used internally by the loader and backends, and useful directly for advanced use cases (custom backends, custom merge strategies).

Merging

yaconfiglib.utils.merge.MergeMethod

Bases: IntEnum

Built-in merge strategies.

Deep = 2 class-attribute instance-attribute

Simple = 1 class-attribute instance-attribute

Substitute = 3 class-attribute instance-attribute

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

Source code in src/yaconfiglib/utils/merge.py

def __call__(
    self,
    a: object,
    b: object,
    *,
    memo: dict | None = None,
    **options,
):
    method: Merge = getattr(self, f"_{self.name.lower()}")
    return method(a, b, memo=memo, **options)

yaconfiglib.utils.merge.Merge

Bases: Protocol

Protocol for any callable that merges two objects.

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

Source code in src/yaconfiglib/utils/merge.py

def __call__(
    self,
    a: object,
    b: object,
    *,
    memo: dict | None = None,
    **options,
) -> object: ...

yaconfiglib.utils.merge.is_scalar(obj)

Source code in src/yaconfiglib/utils/merge.py

def is_scalar(obj: object) -> bool:
    return isinstance(obj, _SCALAR_TYPES)

yaconfiglib.utils.merge.is_array(obj, mutable=False)

Return True if obj is a sequence but not a mapping.

When mutable is True, also require that the sequence supports item assignment (i.e. is a :class:~typing.MutableSequence).

Source code in src/yaconfiglib/utils/merge.py

def is_array(obj, mutable: bool = False) -> bool:
    """Return True if *obj* is a sequence but not a mapping.

    When *mutable* is True, also require that the sequence supports item
    assignment (i.e. is a :class:`~typing.MutableSequence`).
    """
    if isinstance(obj, (list, tuple)):
        return not mutable or isinstance(obj, list)
    if isinstance(obj, (str, bytes, dict)):
        return False
    if isinstance(obj, typing.Mapping):
        return False
    if mutable:
        return isinstance(obj, typing.MutableSequence)
    return isinstance(obj, typing.Sequence)

Jinja2 interpolation

yaconfiglib.utils.jinja2.interpolate(data, globals=None, environment=None)

Recursively interpolate Jinja2 templates within data.

• Strings: rendered as Jinja2 templates. A bare {{ expr }} (no surrounding text) is evaluated as a Python expression so that the return type is preserved (e.g. an integer stays an integer).
• Mappings: keys and values are interpolated recursively.
• Sequences: each element is interpolated recursively.

Returns the interpolated object (may differ in type from data for pure-expression strings).

Source code in src/yaconfiglib/utils/jinja2.py

def interpolate(data: object, globals: dict | None = None, environment: Environment | None = None) -> object:
    """Recursively interpolate Jinja2 templates within *data*.

    * **Strings**: rendered as Jinja2 templates.  A bare ``{{ expr }}``
      (no surrounding text) is evaluated as a Python expression so that
      the return type is preserved (e.g. an integer stays an integer).
    * **Mappings**: keys and values are interpolated recursively.
    * **Sequences**: each element is interpolated recursively.

    Returns the interpolated object (may differ in type from *data* for
    pure-expression strings).
    """
    globals = {} if globals is None else globals

    if isinstance(data, str):
        # Fast path: a string with no Jinja delimiter renders to itself, so
        # skip the cache lookup + Template.render entirely. Most config strings
        # are plain text — this avoids paying Jinja for every one of them.
        if not any(marker in data for marker in _JINJA_MARKERS):
            return data
        stripped = data.strip()
        # Pure Jinja2 expression: {{ expr }} — evaluate to preserve type.
        if stripped.startswith("{{") and stripped.endswith("}}"):
            inner = stripped[2:-2].strip()
            if "{{" not in inner:
                result = eval(inner, environment=environment)(**globals)
                logger.debug("interpolated expression %r -> %r", data, result)
                return result
        result = compile(data, environment=environment)(**globals)
        if result != data:
            logger.debug("interpolated template %r -> %r", data, result)
        return result

    if isinstance(data, _ty.Mapping):
        if not isinstance(data, _ty.MutableMapping):
            data = dict(data)
        for key in list(data.keys()):
            value = data.pop(key)
            new_key = interpolate(key, globals, environment=environment)
            data[new_key] = interpolate(value, globals, environment=environment)
        return data

    if isinstance(data, _ty.Iterable) and not isinstance(data, (str, bytes)):
        if not isinstance(data, _ty.MutableSequence):
            data = list(data)
        for idx, value in enumerate(data):
            data[idx] = interpolate(value, globals, environment=environment)
        return data

    return data

yaconfiglib.utils.jinja2.load_template(source, name=None, filename=None, environment=None, globals=None)

Compile source into a :class:~jinja2.Template.

Source code in src/yaconfiglib/utils/jinja2.py

def load_template(
    source: str,
    name: str | None = None,
    filename: str | None = None,
    environment: Environment | None = None,
    globals: _ty.MutableMapping | None = None,
) -> Template:
    """Compile *source* into a :class:`~jinja2.Template`."""
    env = environment or DEFAULT_ENV
    code = env.compile(source, name, filename)
    return Template.from_code(env, code, env.make_globals(globals))

yaconfiglib.utils.jinja2.compile(code, environment=None, globals=None)

Return a render callable for code (a Jinja2 template string).

Source code in src/yaconfiglib/utils/jinja2.py

def compile(
    code: str,
    environment: Environment | None = None,
    globals: _ty.MutableMapping | None = None,
) -> _ty.Callable[..., str]:
    """Return a render callable for *code* (a Jinja2 template string)."""
    env = environment or DEFAULT_ENV
    cached = _cache_get(_COMPILE_CACHE, code, env)
    if cached is not None:
        return cached
    render = load_template(code, environment=env, globals=globals).render
    _cache_put(_COMPILE_CACHE, code, env, render)
    return render

yaconfiglib.utils.jinja2.eval(code, environment=None, globals=None)

Return a callable that evaluates code as a Jinja2 expression.

The expression result is captured via a {% do %} statement and returned from the callable, preserving non-string Python types.

Source code in src/yaconfiglib/utils/jinja2.py

def eval(
    code: str,
    environment: Environment | None = None,
    globals: _ty.MutableMapping | None = None,
) -> _ty.Callable[..., object]:
    """Return a callable that evaluates *code* as a Jinja2 expression.

    The expression result is captured via a ``{% do %}`` statement and
    returned from the callable, preserving non-string Python types.
    """
    env = environment or DEFAULT_ENV
    cached = _cache_get(_EVAL_CACHE, code, env)
    if cached is not None:
        return cached

    template = load_template(
        "{% do _meta.__setitem__('result', " + code + ") %}",
        environment=env,
        globals=globals,
    )

    def _eval(**kwargs) -> object:
        _meta: dict = {}
        template.render(_meta=_meta, **kwargs)
        res = _meta["result"]
        from jinja2 import Undefined
        if isinstance(res, Undefined):
            str(res)  # Forces UndefinedError if strict
            return None
        return res

    _cache_put(_EVAL_CACHE, code, env, _eval)
    return _eval

Source discovery

yaconfiglib.utils.source.parse_sources(sources, base_dir=None, encoding=None, memo=None, path_factory=None, recursive=None)

Resolve sources into a flat stream of loadable :class:Path-like objects.

Each item in sources may be:

• A file path (string or Path) — resolved against base_dir if relative and not a command URI, and glob-expanded if it contains glob magic characters.
• A command URI (exec://, cmd://, sh://, or a +fmt variant) — passed through unresolved and unexpanded so :class:~yaconfiglib.backends.command.CommandBackend can run it.
• An in-memory document: a string/bytes value starting with the "#!\n" marker, where the first line (after the marker) is treated as a virtual filename and the remainder as its content. The content is materialized to a MemPath (or a real temp file as a fallback) so downstream backends can read it like any other file.
• An open stream (:class:io.IOBase) — read fully and materialized the same way as an in-memory document.
• A nested iterable of any of the above — flattened recursively.

Parameters:

Name	Type	Description	Default
sources	Iterable[SourceLike | Iterable[SourceLike]]	The sources to resolve, as passed to ConfigLoader.load().	required
base_dir	Path	Directory relative file paths are joined against.	None
encoding	str	Text encoding used when decoding bytes markers/content.	None
memo	Iterable[str | Path]	Optional set of already-seen path strings, used to detect and skip duplicate sources across recursive calls; mutated in place. Any iterable is accepted and normalized to a set (O(1) membership; the previous list made duplicate detection O(n²)).	None
path_factory	type[Path]	Constructor used to build a Path from a bare string source.	None
recursive	bool	Whether glob expansion should recurse into subdirectories.	None

Yields:

Name	Type	Description
Resolved	Path	class:Path-like objects, one per concrete source
	Path	(glob patterns may yield zero or many).

Raises:

Type	Description
ValueError	If an item in sources is not a recognized source type.

Source code in src/yaconfiglib/utils/source.py

def parse_sources(
    sources: _ty.Iterable[SourceLike | _ty.Iterable[SourceLike]],
    base_dir: Path = None,
    encoding: str = None,
    memo: _ty.Iterable[str | Path] = None,
    path_factory: type[Path] = None,
    recursive: bool = None,
) -> _ty.Iterator[Path]:
    """Resolve *sources* into a flat stream of loadable :class:`Path`-like objects.

    Each item in *sources* may be:

    * A file path (string or ``Path``) — resolved against *base_dir* if
      relative and not a command URI, and glob-expanded if it contains
      glob magic characters.
    * A command URI (``exec://``, ``cmd://``, ``sh://``, or a ``+fmt``
      variant) — passed through unresolved and unexpanded so
      :class:`~yaconfiglib.backends.command.CommandBackend` can run it.
    * An in-memory document: a string/bytes value starting with the
      ``"#!\\n"`` marker, where the first line (after the marker) is
      treated as a virtual filename and the remainder as its content. The
      content is materialized to a ``MemPath`` (or a real temp file as a
      fallback) so downstream backends can read it like any other file.
    * An open stream (:class:`io.IOBase`) — read fully and materialized
      the same way as an in-memory document.
    * A nested iterable of any of the above — flattened recursively.

    Args:
        sources: The sources to resolve, as passed to ``ConfigLoader.load()``.
        base_dir: Directory relative file paths are joined against.
        encoding: Text encoding used when decoding bytes markers/content.
        memo: Optional set of already-seen path strings, used to detect
            and skip duplicate sources across recursive calls; mutated in
            place. Any iterable is accepted and normalized to a set (O(1)
            membership; the previous list made duplicate detection O(n²)).
        path_factory: Constructor used to build a ``Path`` from a bare
            string source.
        recursive: Whether glob expansion should recurse into
            subdirectories.

    Yields:
        Resolved :class:`Path`-like objects, one per concrete source
        (glob patterns may yield zero or many).

    Raises:
        ValueError: If an item in *sources* is not a recognized source type.
    """
    path_factory = path_factory or Path
    recursive = False if recursive is None else bool(recursive)
    if memo is None:
        memo = set()
    elif not isinstance(memo, set):
        memo = set(memo)
    for source in sources:
        if not source:
            continue
        path_marker = "#!"
        newline = "\n"

        if isinstance(source, bytes):
            path_marker = path_marker.encode(encoding or "utf-8")
            newline = newline.encode(encoding or "utf-8")

        # Handle file streams (in-memory or real)
        if isinstance(source, _io.IOBase):
            content = source.read()
            if MemPath is not None:
                # Unique name + default .yaml suffix so backend auto-detection
                # works for an anonymous stream (YAML is yaconfiglib's default).
                path = MemPath(f"stream-{next(_SOURCE_COUNTER)}.yaml")
                path.parent.mkdir(parents=True, exist_ok=True)
                if isinstance(content, str):
                    path.write_text(content, encoding=encoding)
                else:
                    path.write_bytes(content)
                yield path
            else:
                # Fallback to temp file if MemPath is not available
                yield _materialize_temp(content, encoding, ".yaml")
            continue

        elif isinstance(source, (str, Path, bytes)):
            if isinstance(source, (str, bytes)) and source.startswith(path_marker):
                filename, source = source.split(newline, maxsplit=1)
                logger.debug("loading config doc from memory ...")
                filename = filename.removeprefix(path_marker)
                if isinstance(filename, bytes):
                    filename = filename.decode(encoding or "utf-8")
                if not filename:
                    # Unnamed in-memory docs each get a unique virtual name so
                    # two of them never share (and overwrite) one MemPath. The
                    # ``.yaml`` suffix keeps backend auto-detection working for
                    # a bare ``loads("...")`` (YAML is yaconfiglib's default).
                    filename = f"mem-{next(_SOURCE_COUNTER)}.yaml"
                if MemPath is not None:
                    path = MemPath(filename)
                    path.parent.mkdir(parents=True, exist_ok=True)
                    if isinstance(source, bytes):
                        path.write_bytes(source)
                    else:
                        path.write_text(source, encoding=encoding)
                    yield path
                else:
                    # Fallback to temp file if MemPath is not available
                    yield _materialize_temp(source, encoding, filename)
                continue
            elif isinstance(source, Path):
                is_cmd = bool(_CMD_REGEX.match(str(source)))
                path = source
                if base_dir and not is_cmd:
                    try:
                        path = base_dir / source
                    except TypeError:
                        # base_dir type is incompatible with this source path type — use source as-is.
                        logger.debug(
                            "Cannot join base_dir %r with path %r; using path as-is",
                            base_dir,
                            source,
                        )
            else:
                is_cmd = isinstance(source, str) and bool(_CMD_REGEX.match(source))
                path = path_factory(source)
                if base_dir and not is_cmd:
                    try:
                        path = base_dir / source
                    except (TypeError, ValueError):
                        logger.debug(
                            "Cannot join base_dir %r with %r; using path_factory result",
                            base_dir,
                            source,
                        )
            if not is_cmd:
                memo_key = str(path)
                if memo_key in memo:
                    logger.warning("ignoring duplicated file %s" % path)
                    continue
                memo.add(memo_key)
            if not is_cmd and has_glob_pattern(path):
                # stdlib glob pattern fallback uses glob.glob on string paths
                if hasattr(path, "glob") and HAS_PATHLIB_NEXT:
                    try:
                        yield from path.glob("", recursive=recursive)
                    except TypeError:
                        pattern = path.name
                        parent_dir = path.parent
                        yield from parent_dir.glob(pattern)
                else:
                    # Fallback path traversal
                    # If it's a standard Path, glob is supported: path.glob(pattern)
                    # We need to separate directory from the pattern
                    pattern = path.name
                    parent_dir = path.parent
                    yield from parent_dir.glob(pattern)
            else:
                yield path
        elif isinstance(source, _ty.Iterable):
            yield from parse_sources(
                source,
                memo=memo,
                base_dir=base_dir,
                path_factory=path_factory,
                encoding=encoding,
                recursive=recursive,
            )
        else:
            raise ValueError(
                "unable to handle arg %s of type %s"
                % (
                    source,
                    type(source),
                )
            )

yaconfiglib.utils.source.has_glob_pattern(path)

Check if the given Path contains glob pattern characters.

Source code in src/yaconfiglib/utils/source.py

def has_glob_pattern(path: Path) -> bool:
    """Check if the given Path contains glob pattern characters."""
    if hasattr(path, "has_glob_pattern"):
        return path.has_glob_pattern()
    # Fallback checking path string representation directly (faster than path.parts)
    return _glob.has_magic(str(path))