Loader

The core orchestrator plus the standard-library-style module functions.

yaconfiglib.loader.ConfigLoader(base_dir='', *, encoding=None, path_factory=None, loader_factory=None, recursive=None, key_factory=None, log_level=LogLevel.Warning, interpolate=None, merge=ConfigLoaderMergeMethod.Simple, merge_options=None, ignore_error=False, inject_env=False, strict=False, allow_commands=True, sandbox=False)

Bases: ConfigBackend

Orchestrates loading, merging, and interpolating configuration sources.

A ConfigLoader is the main entrypoint for hiera-like configuration loading: given one or more sources (file paths, glob patterns, command URIs, in-memory strings, or nested lists thereof), it resolves each source to the appropriate :class:~yaconfiglib.backends.base.ConfigBackend, parses it, and merges the results together in order using a configurable :class:ConfigLoaderMergeMethod/:class:~yaconfiglib.utils.merge.Merge strategy. Optionally, the merged result is interpolated with Jinja2 (see :attr:interpolate) and wrapped in a :class:DotAccessibleDict for config.some.nested.key style access.

Most users can use the module-level :func:load/:func:loads helpers, which construct a ConfigLoader for a single call. Construct a ConfigLoader instance directly when you need to reuse the same settings (base directory, merge strategy, interpolation options) across multiple loads.

Configure a reusable loader.

Parameters:

Name	Type	Description	Default
base_dir	str | Path	Directory relative-path sources are resolved against. Accepts a string or Path.	''
encoding	str	Default text encoding for reading sources.	None
path_factory	Callable[[str], Path]	Callable used to build a Path from a bare string source. Defaults to :attr:DEFAULT_PATH_FACTORY.	None
loader_factory	type[ConfigBackend]	Callable (path) -> ConfigBackend instance used to select a backend per source. Defaults to :meth:~yaconfiglib.backends.base.ConfigBackend.get_class_by_path-based dispatch.	None
recursive	bool	Whether glob sources should recurse into subdirectories by default.	None
key_factory	Callable[[Path, object], str]	Callable (path, value) -> str producing the merge key used to track/name each loaded document (e.g. for :attr:ConfigLoaderMergeMethod.Hash). Defaults to the source's filename stem. May also be set per-call as a string attribute name or a "%<jinja-expr>" template.	None
log_level	int | LogLevel	Logging verbosity for this loader's module logger.	Warning
interpolate	bool	If True, run Jinja2 interpolation over the merged result after loading (see :func:yaconfiglib.utils.jinja2.interpolate).	None
merge	ConfigLoaderMergeMethod | Merge	The merge strategy applied between successive sources — a :class:ConfigLoaderMergeMethod or any :class:~yaconfiglib.utils.merge.Merge-compatible callable.	Simple
merge_options	dict[str]	Extra keyword options forwarded to the merge callable on every call (e.g. {"mergelists": True}).	None
ignore_error	_IgnoreError | bool	Either a bool (ignore/re-raise all load errors uniformly) or a predicate (error, **context) -> bool deciding per-error whether to skip and continue.	False
inject_env	bool	If True and interpolate is set, expose os.environ to templates as the env global.	False
strict	bool	If True, undefined Jinja2 variables raise during interpolation instead of rendering as empty.	False
allow_commands	bool	If False, loading a command source (cmd://, exec://, sh://, *+fmt://, or a script-extension file) — including one reached via !include — raises :class:CommandsDisabledError instead of executing it. Set this when loading configuration you do not fully trust. Does not restrict a directly-constructed CommandBackend.	True
sandbox	bool	If True, interpolation runs in Jinja2's SandboxedEnvironment, blocking attribute traversal into Python internals (SSTI). Set this when config values may be untrusted.	False

Source code in src/yaconfiglib/loader.py

def __init__(
    self,
    base_dir: str | Path = "",
    *,
    encoding: str = None,
    path_factory: typing.Callable[[str], Path] = None,
    loader_factory: type[ConfigBackend] = None,
    recursive: bool = None,
    key_factory: typing.Callable[[Path, object], str] = None,
    log_level: int | LogLevel = LogLevel.Warning,
    interpolate: bool = None,
    merge: ConfigLoaderMergeMethod | Merge = ConfigLoaderMergeMethod.Simple,
    merge_options: dict[str] = None,
    ignore_error: _IgnoreError | bool = False,
    inject_env: bool = False,
    strict: bool = False,
    allow_commands: bool = True,
    sandbox: bool = False,
) -> None:
    """Configure a reusable loader.

    Args:
        base_dir: Directory relative-path sources are resolved
            against. Accepts a string or ``Path``.
        encoding: Default text encoding for reading sources.
        path_factory: Callable used to build a ``Path`` from a bare
            string source. Defaults to :attr:`DEFAULT_PATH_FACTORY`.
        loader_factory: Callable ``(path) -> ConfigBackend instance``
            used to select a backend per source. Defaults to
            :meth:`~yaconfiglib.backends.base.ConfigBackend.get_class_by_path`-based
            dispatch.
        recursive: Whether glob sources should recurse into
            subdirectories by default.
        key_factory: Callable ``(path, value) -> str`` producing the
            merge key used to track/name each loaded document (e.g.
            for :attr:`ConfigLoaderMergeMethod.Hash`). Defaults to the
            source's filename stem. May also be set per-call as a
            string attribute name or a ``"%<jinja-expr>"`` template.
        log_level: Logging verbosity for this loader's module logger.
        interpolate: If True, run Jinja2 interpolation over the merged
            result after loading (see :func:`yaconfiglib.utils.jinja2.interpolate`).
        merge: The merge strategy applied between successive sources —
            a :class:`ConfigLoaderMergeMethod` or any
            :class:`~yaconfiglib.utils.merge.Merge`-compatible callable.
        merge_options: Extra keyword options forwarded to the merge
            callable on every call (e.g. ``{"mergelists": True}``).
        ignore_error: Either a bool (ignore/re-raise all load errors
            uniformly) or a predicate ``(error, **context) -> bool``
            deciding per-error whether to skip and continue.
        inject_env: If True and *interpolate* is set, expose
            ``os.environ`` to templates as the ``env`` global.
        strict: If True, undefined Jinja2 variables raise during
            interpolation instead of rendering as empty.
        allow_commands: If False, loading a command source (``cmd://``,
            ``exec://``, ``sh://``, ``*+fmt://``, or a script-extension
            file) — including one reached via ``!include`` — raises
            :class:`CommandsDisabledError` instead of executing it. Set
            this when loading configuration you do not fully trust. Does
            not restrict a directly-constructed ``CommandBackend``.
        sandbox: If True, interpolation runs in Jinja2's
            ``SandboxedEnvironment``, blocking attribute traversal into
            Python internals (SSTI). Set this when config values may be
            untrusted.
    """
    self.allow_commands = bool(allow_commands)
    self.sandbox = bool(sandbox)
    self.merge = (
        merge if isinstance(merge, Merge) else ConfigLoaderMergeMethod(merge)
    )
    self.merge_options = {} if merge_options is None else merge_options
    self.interpolate = False if interpolate is None else bool(interpolate)
    self.inject_env = bool(inject_env)
    self.strict = bool(strict)
    # Stored for introspection only. Library code must never call
    # logger.setLevel() on the module logger: doing it here made every
    # ConfigLoader construction (including the module-import-time
    # DEFAULT_LOADER) mutate global logging state, and two loaders with
    # different levels fought over one logger. Callers configure logging.
    self._log_level = LogLevel(log_level or LogLevel.Warning)
    self.path_factory = path_factory or self.DEFAULT_PATH_FACTORY
    self.base_dir = base_dir or ""
    self.encoding = encoding or self.DEFAULT_ENCODING
    self.recursive = False if recursive is None else recursive
    self.loader_factory = loader_factory or (
        lambda path: ConfigBackend.get_class_by_path(path)()
    )
    self.key_factory = key_factory or (lambda path, value: path.stem)
    self.ignore_error = (
        ignore_error
        if callable(ignore_error)
        else lambda error, *args, **kwargs: bool(ignore_error)
    )

allow_commands = bool(allow_commands) instance-attribute

base_dir property writable

encoding = encoding or self.DEFAULT_ENCODING instance-attribute

ignore_error = ignore_error if callable(ignore_error) else (lambda error, *args, **kwargs: bool(ignore_error)) instance-attribute

inject_env = bool(inject_env) instance-attribute

interpolate = False if interpolate is None else bool(interpolate) instance-attribute

key_factory = key_factory or (lambda path, value: path.stem) instance-attribute

loader_factory = loader_factory or (lambda path: ConfigBackend.get_class_by_path(path)()) instance-attribute

merge = merge if isinstance(merge, Merge) else ConfigLoaderMergeMethod(merge) instance-attribute

merge_options = {} if merge_options is None else merge_options instance-attribute

path_factory = path_factory or self.DEFAULT_PATH_FACTORY instance-attribute

recursive = False if recursive is None else recursive instance-attribute

sandbox = bool(sandbox) instance-attribute

strict = bool(strict) instance-attribute

load(*pathname, recursive=None, encoding=None, loader=None, transform=None, default=None, key_factory=None, flatten=False, interpolate=None, merge=None, merge_options=None, allow_commands=None, sandbox=None, **reader_args)

Load, merge, and (optionally) interpolate one or more configuration sources.

Each item in pathname is resolved via :func:~yaconfiglib.utils.source.parse_sources (expanding globs, nested lists, in-memory #!-marked strings, streams, and command URIs), parsed with the backend selected for it, and merged into the running result in order using merge.

Parameters:

Name	Type	Description	Default
*pathname	SourceLike	One or more sources — file paths, glob patterns, command URIs (cmd://...), in-memory content, open streams, or nested iterables of any of these. If omitted entirely, loads a single empty in-memory document.	()
recursive	bool	Overrides the instance's recursive for glob expansion during this call.	None
encoding	str	Overrides the instance's encoding for this call.	None
loader	str	Backend name, backend instance, or callable selecting the backend for every source loaded in this call, overriding per-source auto-detection.	None
transform	str	A Jinja2 expression string evaluated against each loaded document (as value) before merging, letting you reshape a document inline.	None
default	object	Initial value merged against, used when pathname yields no sources.	None
key_factory	str | Callable[[Path], str]	Overrides the instance's key_factory for this call.	None
flatten	bool	If True, the final merged result (expected to be a mapping-of-mappings or sequence-of-sequences) is flattened one level — useful when each source contributes items to a shared top-level collection instead of being keyed by itself.	False
interpolate	bool	Overrides the instance's interpolate for this call.	None
merge	ConfigLoaderMergeMethod | Merge	Overrides the instance's merge strategy for this call.	None
merge_options	dict[str]	Overrides the instance's merge_options for this call.	None
**reader_args	object	Additional keyword arguments forwarded to each backend's load() (e.g. backend-specific options like json_decoder_options or ini_default_section).	{}

Returns:

Type	Description
object	The merged (and possibly interpolated) result. Dict results
object	are wrapped in :class:DotAccessibleDict.

Source code in src/yaconfiglib/loader.py

def load(
    self,
    *pathname: SourceLike,
    recursive: bool = None,
    encoding: str = None,
    loader: str = None,
    transform: str = None,
    default: object = None,
    key_factory: str | typing.Callable[[Path], str] = None,
    flatten: bool = False,
    interpolate: bool = None,
    merge: ConfigLoaderMergeMethod | Merge = None,
    merge_options: dict[str] = None,
    allow_commands: bool = None,
    sandbox: bool = None,
    **reader_args: object,
) -> object:
    """Load, merge, and (optionally) interpolate one or more configuration sources.

    Each item in *pathname* is resolved via
    :func:`~yaconfiglib.utils.source.parse_sources` (expanding globs,
    nested lists, in-memory ``#!``-marked strings, streams, and command
    URIs), parsed with the backend selected for it, and merged into the
    running result in order using *merge*.

    Args:
        *pathname: One or more sources — file paths, glob patterns,
            command URIs (``cmd://...``), in-memory content, open
            streams, or nested iterables of any of these. If omitted
            entirely, loads a single empty in-memory document.
        recursive: Overrides the instance's *recursive* for glob
            expansion during this call.
        encoding: Overrides the instance's *encoding* for this call.
        loader: Backend name, backend instance, or callable selecting
            the backend for every source loaded in this call,
            overriding per-source auto-detection.
        transform: A Jinja2 expression string evaluated against each
            loaded document (as ``value``) before merging, letting you
            reshape a document inline.
        default: Initial value merged against, used when *pathname*
            yields no sources.
        key_factory: Overrides the instance's *key_factory* for this
            call.
        flatten: If True, the final merged result (expected to be a
            mapping-of-mappings or sequence-of-sequences) is flattened
            one level — useful when each source contributes items to a
            shared top-level collection instead of being keyed by
            itself.
        interpolate: Overrides the instance's *interpolate* for this
            call.
        merge: Overrides the instance's *merge* strategy for this call.
        merge_options: Overrides the instance's *merge_options* for
            this call.
        **reader_args: Additional keyword arguments forwarded to each
            backend's ``load()`` (e.g. backend-specific options like
            ``json_decoder_options`` or ``ini_default_section``).

    Returns:
        The merged (and possibly interpolated) result. Dict results
        are wrapped in :class:`DotAccessibleDict`.
    """
    encoding = encoding or self.encoding
    interpolate = self.interpolate if interpolate is None else interpolate
    sandbox = self.sandbox if sandbox is None else sandbox
    merge = (
        merge
        if isinstance(merge, Merge)
        else (ConfigLoaderMergeMethod(merge) if merge else self.merge)
    )
    if not merge:
        merge = self.merge
    # Per-call override only — must NOT rewrite self.merge_options (doing so
    # made one call's override silently leak into every later load()).
    merge_options = (
        self.merge_options if merge_options is None else merge_options
    )

    results = default
    _join_init = False

    if not pathname:
        pathname = ("#!\n",)

    for path in parse_sources(
        pathname,
        base_dir=self.base_dir,
        encoding=encoding,
        path_factory=self.path_factory,
    ):
        try:
            name, result = self._load(
                path,
                recursive=recursive,
                encoding=encoding,
                loader=loader,
                transform=transform,
                key_factory=key_factory,
                allow_commands=allow_commands,
                **reader_args,
            )
            if _join_init:
                results = merge(
                    results,
                    result,
                    configloaderkey=name,
                    **merge_options,
                )
            else:
                try:
                    results = merge.init(
                        initial=result,
                        configloaderkey=name,
                        **merge_options,
                    )
                except AttributeError:
                    results = result
                _join_init = True
        # Deliberately broad: ``ignore_error`` is a user predicate designed
        # to decide per-error whether to skip ANY load failure (a YAML parse
        # error, a missing file, a backend error...), so narrowing the tuple
        # would break that contract. KeyboardInterrupt/SystemExit are
        # BaseException and already excluded. The error is never swallowed
        # silently — it is handed to the predicate and logged.
        except Exception as error:  # noqa: BLE001 - feeds the ignore_error predicate
            logger.debug("load error for %s: %s", path, error)
            if self.ignore_error(error, path=path, loader=self):
                continue
            raise

    if flatten:
        if isinstance(results, typing.Mapping):
            result = {
                prop: value
                for _key, result in results.items()
                for prop, value in result.items()
            }
        elif is_array(results):
            result = [r for result in results for r in result]
        else:
            raise TypeError(
                "flatten=True requires merged results to be a mapping or sequence"
            )
    else:
        result = results

    if interpolate:
        import os
        custom_env = _get_jinja_env(self.strict, sandbox)

        # Auto-inject env context if requested
        globals_dict = {}
        if isinstance(result, typing.Mapping):
            globals_dict.update(result)
        if self.inject_env:
            globals_dict["env"] = os.environ

        try:
            result = jinja2.interpolate(
                result,
                globals=globals_dict,
                environment=custom_env,
            )
        except Exception as error:  # noqa: BLE001 - feeds the ignore_error predicate
            logger.debug("interpolation error: %s", error)
            if not self.ignore_error(error, result=result, loader=self):
                raise

    # Wrap dict results in a helper class that supports dot-notation
    if isinstance(result, dict):
        result = DotAccessibleDict(result)

    return result

load_all(*pathname, encoding=None, interpolate=None, **reader_args)

Yield each source's parsed (and optionally interpolated) document individually, without merging.

Unlike :meth:load, which merges every source into a single result, this generator yields one value per resolved source — useful when sources represent independent documents rather than layers of the same configuration (e.g. iterating a directory of unrelated config files).

Parameters:

Name	Type	Description	Default
*pathname	Path | Sequence[Path]	Sources to resolve, same semantics as :meth:load.	()
encoding	str	Overrides the instance's encoding for this call.	None
interpolate	bool	Overrides the instance's interpolate for this call; applied independently to each yielded document.	None
**reader_args	object	Additional keyword arguments forwarded to each backend's load().	{}

Yields:

Type	Description
object	Each source's parsed document, with dict results wrapped in
object	class:DotAccessibleDict.

Source code in src/yaconfiglib/loader.py

def load_all(
    self,
    *pathname: Path | typing.Sequence[Path],
    encoding: str = None,
    interpolate: bool = None,
    **reader_args: object,
) -> typing.Iterator[object]:
    """Yield each source's parsed (and optionally interpolated) document individually, without merging.

    Unlike :meth:`load`, which merges every source into a single
    result, this generator yields one value per resolved source —
    useful when sources represent independent documents rather than
    layers of the same configuration (e.g. iterating a directory of
    unrelated config files).

    Args:
        *pathname: Sources to resolve, same semantics as :meth:`load`.
        encoding: Overrides the instance's *encoding* for this call.
        interpolate: Overrides the instance's *interpolate* for this
            call; applied independently to each yielded document.
        **reader_args: Additional keyword arguments forwarded to each
            backend's ``load()``.

    Yields:
        Each source's parsed document, with dict results wrapped in
        :class:`DotAccessibleDict`.
    """
    interpolate = self.interpolate if interpolate is None else interpolate
    encoding = encoding or self.encoding
    custom_env = _get_jinja_env(self.strict, self.sandbox) if interpolate else None
    for path in parse_sources(
        pathname,
        base_dir=self.base_dir,
        encoding=encoding,
        path_factory=self.path_factory,
    ):
        value = None
        try:
            key, value = self._load(
                path,
                encoding=encoding,
                **reader_args,
            )
            if interpolate:
                globals_dict = {}
                if isinstance(value, typing.Mapping):
                    globals_dict.update(value)
                if self.inject_env:
                    import os
                    globals_dict["env"] = os.environ
                value = jinja2.interpolate(value, globals_dict, environment=custom_env)
            if isinstance(value, dict):
                value = DotAccessibleDict(value)
            yield value

        except Exception as error:  # noqa: BLE001 - feeds the ignore_error predicate
            logger.debug("load_all error for %s: %s", path, error)
            if not self.ignore_error(
                error, path=path, value=value, loader=self
            ):
                raise

load_as(model_cls, *pathname, **kwargs)

Load configuration sources and instantiate as model_cls.

Supports Pydantic models (if installed) or dataclasses. If neither matches, falls back to passing kwargs/dict unpacking to the constructor.

Source code in src/yaconfiglib/loader.py

def load_as(self, model_cls: type[T], *pathname: SourceLike, **kwargs) -> T:
    """Load configuration sources and instantiate as *model_cls*.

    Supports Pydantic models (if installed) or dataclasses. If neither matches,
    falls back to passing kwargs/dict unpacking to the constructor.
    """
    data = self.load(*pathname, **kwargs)
    if not isinstance(data, dict):
        raise TypeError("Loaded configuration must be a dictionary to load as a model")

    # Try Pydantic integration (strictly optional)
    try:
        import pydantic
        if issubclass(model_cls, pydantic.BaseModel):
            # Pydantic V2 and V1 compatibility helper
            if hasattr(model_cls, "model_validate"):
                return model_cls.model_validate(data)
            elif hasattr(model_cls, "parse_obj"):
                return model_cls.parse_obj(data)
    except ImportError:
        pass

    # Try dataclass
    from dataclasses import is_dataclass
    if is_dataclass(model_cls):
        # Safe init passing only valid dataclass field names
        import inspect
        sig = inspect.signature(model_cls.__init__)
        valid_keys = {name for name, param in sig.parameters.items() if param.kind in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY)}
        filtered = {k: v for k, v in data.items() if k in valid_keys}
        return model_cls(**filtered)

    return model_cls(**data)

yaconfiglib.loader.DotAccessibleDict

Bases: dict

Dictionary subclass supporting dot-notation queries and attribute access.

__getattr__(name)

Source code in src/yaconfiglib/loader.py

def __getattr__(self, name: str) -> object:
    try:
        val = self[name]
        if isinstance(val, dict) and not isinstance(val, DotAccessibleDict):
            val = DotAccessibleDict(val)
            self[name] = val
        return val
    except KeyError:
        raise AttributeError(f"'DotAccessibleDict' object has no attribute '{name}'")

__setattr__(name, value)

Source code in src/yaconfiglib/loader.py

def __setattr__(self, name: str, value: object) -> None:
    self[name] = value

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

Support dot-notation traversal, e.g., get("database.credentials.user", dig=True).

Source code in src/yaconfiglib/loader.py

def get(self, key: str, default: object = None, dig: bool = True) -> object:
    """Support dot-notation traversal, e.g., get("database.credentials.user", dig=True)."""
    if key in self:
        val = super().get(key, default)
        if isinstance(val, dict) and not isinstance(val, DotAccessibleDict):
            val = DotAccessibleDict(val)
            self[key] = val
        return val

    if dig and "." in key:
        parts = key.split(".")
        current = self
        for part in parts:
            if not isinstance(current, dict):
                return default
            parent = current
            try:
                current = current[part]
            except KeyError:
                return default
            if current is None:
                return default
            if isinstance(current, dict) and not isinstance(current, DotAccessibleDict):
                current = DotAccessibleDict(current)
                parent[part] = current
        return current
    val = super().get(key, default)
    if isinstance(val, dict) and not isinstance(val, DotAccessibleDict):
        val = DotAccessibleDict(val)
        self[key] = val
    return val

yaconfiglib.loader.ConfigLoaderMergeMethod

Bases: _ConfigLoaderMergeMethod, MergeMethod, Protocol

Module-level functions

yaconfiglib.loader.load(fp, **kwargs)

Load configuration from a file pointer or file path.

Source code in src/yaconfiglib/loader.py

def load(fp: typing.Any, **kwargs) -> object:
    """Load configuration from a file pointer or file path."""
    load_keys = {"recursive", "encoding", "loader", "transform", "default", "key_factory", "flatten", "interpolate", "merge", "merge_options", "master"}
    loader_kwargs = {k: v for k, v in kwargs.items() if k not in load_keys}
    load_kwargs = {k: v for k, v in kwargs.items() if k in load_keys}
    loader_inst = ConfigLoader(**loader_kwargs)
    return loader_inst.load(fp, **load_kwargs)

yaconfiglib.loader.loads(s, **kwargs)

Load configuration from a string or bytes in memory.

Source code in src/yaconfiglib/loader.py

def loads(s: str | bytes, **kwargs) -> object:
    """Load configuration from a string or bytes in memory."""
    load_keys = {"recursive", "encoding", "loader", "transform", "default", "key_factory", "flatten", "interpolate", "merge", "merge_options", "master"}
    loader_kwargs = {k: v for k, v in kwargs.items() if k not in load_keys}
    load_kwargs = {k: v for k, v in kwargs.items() if k in load_keys}
    loader_inst = ConfigLoader(**loader_kwargs)
    # Use parse_sources inline memory doc marker
    marker = "#!\n"
    if isinstance(s, bytes):
        content = marker.encode("utf-8") + s
    else:
        content = marker + s
    return loader_inst.load(content, **load_kwargs)

yaconfiglib.loader.dump(obj, fp, **kwargs)

Dump configuration object to a file pointer or file path.

Source code in src/yaconfiglib/loader.py

def dump(obj: object, fp: typing.Any, **kwargs) -> None:
    """Dump configuration object to a file pointer or file path."""
    content = dumps(obj, **kwargs)
    if hasattr(fp, "write"):
        fp.write(content)
    else:
        with open(fp, "w", encoding="utf-8") as f:
            f.write(content)

yaconfiglib.loader.dumps(obj, **kwargs)

Dump configuration object to string (delegates to YamlConfig dumper by default).

Source code in src/yaconfiglib/loader.py

def dumps(obj: object, **kwargs) -> str:
    """Dump configuration object to string (delegates to YamlConfig dumper by default)."""
    from .backends.yaml import YamlConfig
    backend = YamlConfig()
    return backend.dumps(obj, **kwargs)