hyera.backends

Data backends: a self-registering Backend registry.

Every format or provider is a :class:Backend subclass. Registration is by subclassing (NAMES, keyed by namespace) rather than an explicit call; lookup goes through :meth:Backend.find/:meth:Backend.get/:meth:Backend.new.

SOPS_TIMEOUT = 30 module-attribute

Backend(conf=None, *, strict=None)

A data format and/or Hiera 5 provider, registered by name.

A subclass registers itself by declaring NAMES, a mapping of namespace (one of :attr:KINDS) to the names (plain strings and/or :class:NamePattern) it answers to in that namespace. Puppet function names (data_hash/lookup_key/data_dig values in a v5 hierarchy), v3 backend names, plain format names and CLI/render names are separate namespaces -- a name is only ever looked up within one.

Anything a backend does not implement raises :class:NotImplementedError from the methods below (the base class's defaults); callers turn that into a specific, user-facing error.

Parameters:

Name	Type	Description	Default
conf	Optional[Mapping[str, Any]]	the hierarchy entry's/defaults's own mapping (never used by the base class; a subclass may read its own keys from it).	None
strict	Optional[Union[Strict, str]]	overrides the call-time default (see :attr:strict).	None

Source code in src/hyera/backends/_base.py

def __init__(
    self,
    conf: _ty.Optional[_ty.Mapping[str, _ty.Any]] = None,
    *,
    strict: _ty.Optional[_ty.Union[Strict, str]] = None,
) -> None:
    if conf is not None and not isinstance(conf, _abc.Mapping):
        raise TypeError(
            "conf must be a mapping or None, not {}".format(type(conf).__name__)
        )
    self.conf: _ty.Mapping[str, _ty.Any] = conf or {}
    if strict is not None and strict not in _STRICT_VALUES:
        raise ValueError(
            "strict must be one of {!r}, not {!r}".format(_STRICT_VALUES, strict)
        )
    self._strict = _plain(strict)
    self.name: _ty.Optional[str] = type(self)._default_name()

EXTENSIONS = () class-attribute

KINDS = ('function', 'v3', 'format', 'render') class-attribute

NAMES = {} class-attribute

conf = conf or {} instance-attribute

limits property

The :class:~hyera.Limits of the read in progress, or None when nothing is bounded. A backend that parses untrusted text honours the fields it can.

name = type(self)._default_name() instance-attribute

strict property

This backend's effective strictness: an explicit constructor value, else the call-time default (see :func:_default_strict). Never cache a result that depends on this -- it can change call to call once the ContextVar-backed default lands.

__init_subclass__(**kwargs)

Source code in src/hyera/backends/_base.py

def __init_subclass__(cls, **kwargs: _ty.Any) -> None:
    super().__init_subclass__(**kwargs)
    names = cls.__dict__.get("NAMES", {})
    if not names:
        return
    for kind in names:
        if kind not in Backend.KINDS:
            raise ValueError(
                "{}: unknown backend kind {!r}; known: {}".format(
                    cls.__name__, kind, Backend.KINDS
                )
            )
    if "render" in names and "dumps" not in cls.__dict__:
        raise TypeError(
            "{}: a 'render' backend must override dumps()".format(cls.__name__)
        )
    if "format" in names and "loads" not in cls.__dict__:
        raise TypeError(
            "{}: a 'format' backend must override loads()".format(cls.__name__)
        )
    for kind, entries in names.items():
        registry = Backend._REGISTRY.setdefault(kind, {"exact": {}, "patterns": []})
        for entry in entries:
            if isinstance(entry, NamePattern):
                for existing, other in registry["patterns"]:
                    if existing.display == entry.display:
                        raise ValueError(
                            "{!r} ({} kind) is already registered to {}; "
                            "{} cannot reuse it".format(
                                entry.display, kind, other.__name__, cls.__name__
                            )
                        )
                for exact, other in registry["exact"].items():
                    if entry.regex.fullmatch(exact):
                        raise ValueError(
                            "pattern {!r} ({} kind) would take over the "
                            "name {!r} registered to {}; {} cannot "
                            "register it".format(
                                entry.display,
                                kind,
                                exact,
                                other.__name__,
                                cls.__name__,
                            )
                        )
                registry["patterns"].append((entry, cls))
            else:
                other = registry["exact"].get(entry)
                if other is not None:
                    raise ValueError(
                        "{!r} ({} kind) is already registered to {}; "
                        "{} cannot reuse it".format(
                            entry, kind, other.__name__, cls.__name__
                        )
                    )
                for pattern, other in registry["patterns"]:
                    if pattern.regex.fullmatch(entry):
                        raise ValueError(
                            "{!r} ({} kind) is already answered by the "
                            "pattern {!r} of {}; {} cannot register "
                            "it".format(
                                entry,
                                kind,
                                pattern.display,
                                other.__name__,
                                cls.__name__,
                            )
                        )
                registry["exact"][entry] = cls

check_available() classmethod

Raise :class:BackendError if this backend cannot be used (e.g. a missing optional dependency). A no-op by default.

Source code in src/hyera/backends/_base.py

@classmethod
def check_available(cls) -> None:
    """Raise :class:`BackendError` if this backend cannot be used (e.g.
    a missing optional dependency). A no-op by default."""

data_dig(key_segments, options, context)

The data_dig provider hook: resolve one already-split key_segments path in one hierarchy location. Raises :class:NotImplementedError unless overridden.

Parameters:

Name	Type	Description	Default
key_segments	Sequence[str]	the already-split dotted key path.	required
options	Mapping[str, Any]	the hierarchy entry's options.	required
context	LookupContext	the per-location :class:LookupContext.	required

Returns:

Type	Description
Any	the found value.

Raises:

Type	Description
NotImplementedError	the base class; a subclass must override this to support data_dig.

Source code in src/hyera/backends/_base.py

def data_dig(
    self,
    key_segments: _ty.Sequence[str],
    options: _ty.Mapping[str, _ty.Any],
    context: LookupContext,
) -> _ty.Any:
    """The ``data_dig`` provider hook: resolve one already-split
    ``key_segments`` path in one hierarchy location. Raises
    :class:`NotImplementedError` unless overridden.

    :param key_segments: the already-split dotted key path.
    :param options: the hierarchy entry's ``options``.
    :param context: the per-location :class:`LookupContext`.
    :returns: the found value.
    :raises NotImplementedError: the base class; a subclass must
        override this to support ``data_dig``.
    """
    raise NotImplementedError(
        "{} does not implement .data_dig()".format(type(self).__name__)
    )

data_hash(path, options, context)

The data_hash provider hook: parse the whole file at path and adapt it into hiera data (see :meth:_as_data_hash).

The base implementation is a file function: it accepts only a single path location and no hierarchy options (:meth:_require_path_only), Puppet's own yaml_data/ json_data/hocon_data contract.

Parameters:

Name	Type	Description	Default
path	'Path'	the location's file path.	required
options	Mapping[str, Any]	the hierarchy entry's options.	required
context	LookupContext	the per-location :class:LookupContext; calling its not_found() is an error here.	required

Returns:

Type	Description
Dict[str, Any]	the parsed data, adapted into a hash.

Raises:

Type	Description
ConfigError	options carries anything besides path.
BackendError	the file could not be read or parsed.

Source code in src/hyera/backends/_base.py

def data_hash(
    self,
    path: "Path",
    options: _ty.Mapping[str, _ty.Any],
    context: LookupContext,
) -> _ty.Dict[str, _ty.Any]:
    """The ``data_hash`` provider hook: parse the whole file at
    ``path`` and adapt it into hiera data (see :meth:`_as_data_hash`).

    The base implementation is a *file* function: it accepts only a
    single ``path`` location and no hierarchy ``options``
    (:meth:`_require_path_only`), Puppet's own ``yaml_data``/
    ``json_data``/``hocon_data`` contract.

    :param path: the location's file path.
    :param options: the hierarchy entry's ``options``.
    :param context: the per-location :class:`LookupContext`; calling its
        ``not_found()`` is an error here.
    :returns: the parsed data, adapted into a hash.
    :raises ConfigError: ``options`` carries anything besides ``path``.
    :raises BackendError: the file could not be read or parsed.
    """
    self._require_path_only(path, options)
    return self._as_data_hash(self.load(path), path)

dump(obj, fp, **kw)

Render obj and write it to the open text file fp.

Parameters:

Name	Type	Description	Default
obj	Any	the value to render.	required
fp	'_ty.IO[str]'	the open text file to write to.	required
kw	Any	format-specific rendering options.	{}

Source code in src/hyera/backends/_base.py

def dump(self, obj: _ty.Any, fp: "_ty.IO[str]", **kw: _ty.Any) -> None:
    """Render ``obj`` and write it to the open text file ``fp``.

    :param obj: the value to render.
    :param fp: the open text file to write to.
    :param kw: format-specific rendering options.
    """
    fp.write(self.dumps(obj, **kw))

dumps(obj, **kw)

Render obj back to text. Raises path-free problem text.

Parameters:

Name	Type	Description	Default
obj	Any	the value to render.	required
kw	Any	format-specific rendering options.	{}

Returns:

Type	Description
str	the rendered text.

Raises:

Type	Description
NotImplementedError	the base class; a subclass must override this to support render rendering.

Source code in src/hyera/backends/_base.py

def dumps(self, obj: _ty.Any, **kw: _ty.Any) -> str:
    """Render ``obj`` back to text. Raises path-free problem text.

    :param obj: the value to render.
    :param kw: format-specific rendering options.
    :returns: the rendered text.
    :raises NotImplementedError: the base class; a subclass must
        override this to support ``render`` rendering.
    """
    raise NotImplementedError(
        "{} does not implement .dumps()".format(type(self).__name__)
    )

find(name, kind='function') classmethod

The registered class for name in kind, or None.

Parameters:

Name	Type	Description	Default
name	str	the registered name (or a matching pattern) to find.	required
kind	Union[BackendKind, str]	the namespace to search.	'function'

Returns:

Type	Description
'_ty.Optional[_ty.Type[Backend]]'	the class, or None when unregistered.

Raises:

Type	Description
TypeError	name is not a str, or kind is neither a :class:BackendKind nor a str.
ValueError	kind names no namespace.

Source code in src/hyera/backends/_base.py

@classmethod
def find(
    cls, name: str, kind: _ty.Union[BackendKind, str] = "function"
) -> "_ty.Optional[_ty.Type[Backend]]":
    """The registered class for ``name`` in ``kind``, or ``None``.

    :param name: the registered name (or a matching pattern) to find.
    :param kind: the namespace to search.
    :returns: the class, or ``None`` when unregistered.
    :raises TypeError: ``name`` is not a ``str``, or ``kind`` is neither a
        :class:`BackendKind` nor a ``str``.
    :raises ValueError: ``kind`` names no namespace.
    """
    found, _captures = cls._match(name, kind)
    return found

for_path(path) classmethod

The format-namespace backend class whose :attr:EXTENSIONS has the longest case-sensitive suffix match against path, or None.

Parameters:

Name	Type	Description	Default
path	Union[str, 'os.PathLike[str]']	the file path to match.	required

Returns:

Type	Description
'_ty.Optional[_ty.Type[Backend]]'	the class, or None when nothing matches.

Source code in src/hyera/backends/_base.py

@classmethod
def for_path(
    cls, path: _ty.Union[str, "os.PathLike[str]"]
) -> "_ty.Optional[_ty.Type[Backend]]":
    """The ``format``-namespace backend class whose :attr:`EXTENSIONS`
    has the longest case-sensitive suffix match against ``path``, or
    ``None``.

    :param path: the file path to match.
    :returns: the class, or ``None`` when nothing matches.
    """
    _load_entry_points()
    name = os.fspath(path)
    registry = cls._REGISTRY.get("format", {"exact": {}, "patterns": []})
    candidates = {klass for klass in registry["exact"].values()}
    candidates.update(klass for _pattern, klass in registry["patterns"])
    best_cls = None
    best_len = -1
    for candidate in candidates:
        for ext in candidate.EXTENSIONS:
            if name.endswith(ext) and len(ext) > best_len:
                best_len = len(ext)
                best_cls = candidate
    return best_cls

get(name, kind='function') classmethod

The registered, available class for name in kind.

Raises :class:ValueError for an unknown name (listing the known names) and, via :meth:check_available, :class:BackendError for a registered backend whose optional dependency is missing.

Parameters:

Name	Type	Description	Default
name	str	the registered name (or a matching pattern) to get.	required
kind	Union[BackendKind, str]	the namespace to search.	'function'

Returns:

Type	Description
'_ty.Type[Backend]'	the class.

Raises:

Type	Description
TypeError	name is not a str, or kind is neither a :class:BackendKind nor a str.
ValueError	name is unregistered in kind, or kind names no namespace.
BackendError	name is registered but unusable (a missing optional dependency).

Source code in src/hyera/backends/_base.py

@classmethod
def get(
    cls, name: str, kind: _ty.Union[BackendKind, str] = "function"
) -> "_ty.Type[Backend]":
    """The registered, available class for ``name`` in ``kind``.

    Raises :class:`ValueError` for an unknown name (listing the known
    names) and, via :meth:`check_available`, :class:`BackendError` for a
    registered backend whose optional dependency is missing.

    :param name: the registered name (or a matching pattern) to get.
    :param kind: the namespace to search.
    :returns: the class.
    :raises TypeError: ``name`` is not a ``str``, or ``kind`` is neither a
        :class:`BackendKind` nor a ``str``.
    :raises ValueError: ``name`` is unregistered in ``kind``, or ``kind``
        names no namespace.
    :raises BackendError: ``name`` is registered but unusable (a missing
        optional dependency).
    """
    kind = check_kind(kind, cls.KINDS)
    found = cls.find(name, kind)
    if found is None:
        raise unknown_backend(kind, name, cls.names(kind))
    found.check_available()
    return found

implements(op) classmethod

Whether this class overrides what op needs.

load/dump follow loads/dumps; data_hash is true when data_hash or loads is overridden (the base data_hash delegates to load -> loads).

Parameters:

Name	Type	Description	Default
op	str	one of "load", "dump", "data_hash", "loads", "dumps", "lookup_key", "data_dig".	required

Returns:

Type	Description
bool	whether this class implements op.

Raises:

Type	Description
ValueError	op is not one of those names.

Source code in src/hyera/backends/_base.py

@classmethod
def implements(cls, op: str) -> bool:
    """Whether this class overrides what ``op`` needs.

    ``load``/``dump`` follow ``loads``/``dumps``; ``data_hash`` is true
    when ``data_hash`` or ``loads`` is overridden (the base
    ``data_hash`` delegates to ``load`` -> ``loads``).

    :param op: one of ``"load"``, ``"dump"``, ``"data_hash"``,
        ``"loads"``, ``"dumps"``, ``"lookup_key"``, ``"data_dig"``.
    :returns: whether this class implements ``op``.
    :raises ValueError: ``op`` is not one of those names.
    """
    if op == "load":
        return cls._overrides("loads") or cls._overrides("load")
    if op == "dump":
        return cls._overrides("dumps") or cls._overrides("dump")
    if op == "data_hash":
        return cls._overrides("data_hash") or cls._overrides("loads")
    if op in ("loads", "dumps", "lookup_key", "data_dig"):
        return cls._overrides(op)
    raise ValueError("unknown Backend operation {!r}".format(op))

load(source)

Parse source -- a path-like or a file object.

Reads the bytes and decodes them as strict UTF-8 (Puppet's data files are read this way: pops/lookup/context.rb:53); a str already produced by a text file object is used as-is. A decode error, or the problem text of a :class:BackendError from :meth:loads, is re-raised (outside the except block, so neither holds the original as __cause__/__context__) as BackendError("Unable to parse (<path>): <problem>", path=...). Because .path is set here, a caller (_LocationStore.load_file) does not need to prefix it again.

Parameters:

Name	Type	Description	Default
source	Union[str, 'os.PathLike[str]', '_ty.IO[str]', '_ty.IO[bytes]']	a path-like, or an already-open file object.	required

Returns:

Type	Description
Any	the parsed value.

Raises:

Type	Description
BackendError	source could not be read, decoded as UTF-8 or parsed by :meth:loads; an unreadable source reads Unable to read (<path>): <reason>.

Source code in src/hyera/backends/_base.py

def load(
    self,
    source: _ty.Union[str, "os.PathLike[str]", "_ty.IO[str]", "_ty.IO[bytes]"],
) -> _ty.Any:
    """Parse ``source`` -- a path-like or a file object.

    Reads the bytes and decodes them as strict UTF-8 (Puppet's data
    files are read this way: ``pops/lookup/context.rb:53``); a ``str``
    already produced by a text file object is used as-is. A decode
    error, or the problem text of a :class:`BackendError` from
    :meth:`loads`, is re-raised (outside the ``except`` block, so
    neither holds the original as ``__cause__``/``__context__``) as
    ``BackendError("Unable to parse (<path>): <problem>", path=...)``.
    Because ``.path`` is set here, a caller (``_LocationStore.load_file``) does
    not need to prefix it again.

    :param source: a path-like, or an already-open file object.
    :returns: the parsed value.
    :raises BackendError: ``source`` could not be read, decoded as UTF-8 or
        parsed by :meth:`loads`; an unreadable ``source`` reads
        ``Unable to read (<path>): <reason>``.
    """
    is_file_obj = hasattr(source, "read")
    path = getattr(source, "name", "<unknown>") if is_file_obj else source
    problem = None
    try:
        if is_file_obj:
            content = source.read()
            text = content if isinstance(content, str) else content.decode("utf-8")
        else:
            reader = getattr(source, "read_bytes", None)
            if reader is not None:
                data = reader()
            else:
                with open(os.fspath(source), "rb") as fh:
                    data = fh.read()
            text = data.decode("utf-8")
        return self.loads(text)
    except UnicodeDecodeError as e:
        problem = str(e)
    except BackendError as e:
        problem = str(e)
    except OSError as e:
        raise BackendError(
            "Unable to read ({}): {}".format(path, e.strerror or e), path=str(path)
        ) from e
    raise BackendError(
        "Unable to parse ({}): {}".format(path, problem), path=str(path)
    )

loads(text)

Parse text (a str). Raises path-free problem text.

Parameters:

Name	Type	Description	Default
text	str	the text to parse.	required

Returns:

Type	Description
Any	the parsed value.

Raises:

Type	Description
NotImplementedError	the base class; a subclass must override this to support format/function parsing.

Source code in src/hyera/backends/_base.py

def loads(self, text: str) -> _ty.Any:
    """Parse ``text`` (a ``str``). Raises path-free problem text.

    :param text: the text to parse.
    :returns: the parsed value.
    :raises NotImplementedError: the base class; a subclass must
        override this to support ``format``/``function`` parsing.
    """
    raise NotImplementedError(
        "{} does not implement .loads()".format(type(self).__name__)
    )

lookup_key(key, options, context)

The lookup_key provider hook: resolve one dotted key in one hierarchy location. Raises :class:NotImplementedError unless overridden (see :class:EyamlBackend).

Parameters:

Name	Type	Description	Default
key	str	the dotted key to resolve.	required
options	Mapping[str, Any]	the hierarchy entry's options.	required
context	LookupContext	the per-location :class:LookupContext.	required

Returns:

Type	Description
Any	the found value.

Raises:

Type	Description
NotImplementedError	the base class; a subclass must override this to support lookup_key.

Source code in src/hyera/backends/_base.py

def lookup_key(
    self,
    key: str,
    options: _ty.Mapping[str, _ty.Any],
    context: LookupContext,
) -> _ty.Any:
    """The ``lookup_key`` provider hook: resolve one dotted ``key`` in
    one hierarchy location. Raises :class:`NotImplementedError` unless
    overridden (see :class:`EyamlBackend`).

    :param key: the dotted key to resolve.
    :param options: the hierarchy entry's ``options``.
    :param context: the per-location :class:`LookupContext`.
    :returns: the found value.
    :raises NotImplementedError: the base class; a subclass must
        override this to support ``lookup_key``.
    """
    raise NotImplementedError(
        "{} does not implement .lookup_key()".format(type(self).__name__)
    )

names(kind='function') classmethod

Registered names in kind: exact names in registration order, then patterns by their :attr:NamePattern.display.

Parameters:

Name	Type	Description	Default
kind	Union[BackendKind, str]	the namespace to list.	'function'

Returns:

Type	Description
List[str]	the registered names.

Raises:

Type	Description
TypeError	kind is neither a :class:BackendKind nor a str.
ValueError	kind names no namespace.

Source code in src/hyera/backends/_base.py

@classmethod
def names(cls, kind: _ty.Union[BackendKind, str] = "function") -> _ty.List[str]:
    """Registered names in ``kind``: exact names in registration order,
    then patterns by their :attr:`NamePattern.display`.

    :param kind: the namespace to list.
    :returns: the registered names.
    :raises TypeError: ``kind`` is neither a :class:`BackendKind` nor a ``str``.
    :raises ValueError: ``kind`` names no namespace.
    """
    _load_entry_points()
    kind = check_kind(kind, cls.KINDS)
    registry = cls._REGISTRY[kind]
    return list(registry["exact"].keys()) + [
        pattern.display for pattern, _klass in registry["patterns"]
    ]

new(name, conf=None, *, kind='function', strict=None) classmethod

Instantiate the registered backend for name in kind.

Any named groups captured by a matching :class:NamePattern are passed as constructor keywords. .name is set to name (the name actually asked for, which may be a pattern instance such as sops_json, not the class's default name).

Parameters:

Name	Type	Description	Default
name	str	the registered name (or a matching pattern) to instantiate.	required
conf	Optional[Mapping[str, Any]]	passed to the backend's constructor.	None
kind	Union[BackendKind, str]	the namespace to search.	'function'
strict	Optional[Union[Strict, str]]	passed to the backend's constructor.	None

Returns:

Type	Description
'Backend'	the new instance.

Raises:

Type	Description
TypeError	name is not a str, kind is neither a :class:BackendKind nor a str, or conf is not a mapping.
ValueError	name is unregistered in kind, kind names no namespace, or strict is not a strictness.
BackendError	name is registered but unusable (a missing optional dependency).

Source code in src/hyera/backends/_base.py

@classmethod
def new(
    cls,
    name: str,
    conf: _ty.Optional[_ty.Mapping[str, _ty.Any]] = None,
    *,
    kind: _ty.Union[BackendKind, str] = "function",
    strict: _ty.Optional[_ty.Union[Strict, str]] = None,
) -> "Backend":
    """Instantiate the registered backend for ``name`` in ``kind``.

    Any named groups captured by a matching :class:`NamePattern` are
    passed as constructor keywords. ``.name`` is set to ``name`` (the
    name actually asked for, which may be a pattern instance such as
    ``sops_json``, not the class's default name).

    :param name: the registered name (or a matching pattern) to
        instantiate.
    :param conf: passed to the backend's constructor.
    :param kind: the namespace to search.
    :param strict: passed to the backend's constructor.
    :returns: the new instance.
    :raises TypeError: ``name`` is not a ``str``, ``kind`` is neither a
        :class:`BackendKind` nor a ``str``, or ``conf`` is not a mapping.
    :raises ValueError: ``name`` is unregistered in ``kind``, ``kind``
        names no namespace, or ``strict`` is not a strictness.
    :raises BackendError: ``name`` is registered but unusable (a missing
        optional dependency).
    """
    kind = check_kind(kind, cls.KINDS)
    found, captures = cls._match(name, kind)
    if found is None:
        raise unknown_backend(kind, name, cls.names(kind))
    found.check_available()
    instance = found(conf, strict=strict, **captures)
    instance.name = name
    return instance

BackendError(*args, path=None)

Bases: HieraError, ValueError

A data file could not be read or parsed. .path names it.

Also a :class:ValueError.

Source code in src/hyera/exceptions.py

def __init__(self, *args: object, path: _ty.Optional[str] = None) -> None:
    super().__init__(*args)
    self.path: _ty.Optional[str] = path

BackendKind

Bases: _StrEnum

Which of :attr:Backend.KINDS a registered name belongs to: the kind= argument of :meth:Backend.find/:meth:Backend.get/ :meth:Backend.new/:meth:Backend.names. A name is only ever registered, and looked up, within one namespace.

FORMAT = 'format' class-attribute instance-attribute

A plain data-format name, also matched by :meth:Backend.for_path's file-extension lookup (yaml, json, ...).

FUNCTION = 'function' class-attribute instance-attribute

A Hiera 5 hierarchy entry's data_hash/lookup_key/ data_dig function name (yaml_data, json_data, ...).

RENDER = 'render' class-attribute instance-attribute

A puppet lookup --render-as output format (s, json, yaml).

V3 = 'v3' class-attribute instance-attribute

A Hiera 3/hiera3_backend backend name (yaml, json, ...).

DotenvBackend(conf=None, *, strict=None)

Bases: Backend

dotenv, in exactly the shape the sops CLI's writer emits it (stores/dotenv/store.go) -- reachable only through :class:SopsBackend.

Source code in src/hyera/backends/_base.py

def __init__(
    self,
    conf: _ty.Optional[_ty.Mapping[str, _ty.Any]] = None,
    *,
    strict: _ty.Optional[_ty.Union[Strict, str]] = None,
) -> None:
    if conf is not None and not isinstance(conf, _abc.Mapping):
        raise TypeError(
            "conf must be a mapping or None, not {}".format(type(conf).__name__)
        )
    self.conf: _ty.Mapping[str, _ty.Any] = conf or {}
    if strict is not None and strict not in _STRICT_VALUES:
        raise ValueError(
            "strict must be one of {!r}, not {!r}".format(_STRICT_VALUES, strict)
        )
    self._strict = _plain(strict)
    self.name: _ty.Optional[str] = type(self)._default_name()

EXTENSIONS = ('.env',) class-attribute

NAMES = {'format': ('dotenv',)} class-attribute

loads(text)

Parse dotenv the way sops's own writer emits it: KEY=value lines, # comments, blank lines skipped, \n unescaped.

Parameters:

Name	Type	Description	Default
text	str	the dotenv text to parse.	required

Returns:

Type	Description
Dict[str, str]	the parsed key/value pairs.

Raises:

Type	Description
BackendError	a non-blank, non-comment line has no =.

Source code in src/hyera/backends/_sops.py

def loads(self, text: str) -> _ty.Dict[str, str]:
    """Parse dotenv the way sops's own writer emits it: ``KEY=value``
    lines, ``#`` comments, blank lines skipped, ``\\n`` unescaped.

    :param text: the dotenv text to parse.
    :returns: the parsed key/value pairs.
    :raises BackendError: a non-blank, non-comment line has no ``=``.
    """
    result: _ty.Dict[str, str] = {}
    for lineno, line in enumerate(text.split("\n"), start=1):
        if line == "" or line.startswith("#"):
            continue
        if "=" not in line:
            raise BackendError("invalid dotenv line {}".format(lineno))
        key, _sep, value = line.partition("=")
        result[key] = value.replace("\\n", "\n")
    return result

EyamlBackend(conf=None, *, strict=None)

Bases: Backend

Puppet's hiera-eyaml lookup_key function, PKCS7 only (behind the optional eyaml extra). Ports functions/eyaml_lookup_key. rb:25-79: the raw .eyaml file loads once per change (through :meth:~hyera._lookup.function_provider.LookupContext.cached_file_data, its raw parse only -- caching the non-Hash rule's strict-sensitive result would freeze whichever strictness read it first, exactly the trap _LocationStore.load_file guards against for data_hash), then each requested key's value is decrypted (:func:hyera.backends._eyaml.decrypt_string); the engine keeps that result until the file changes. The raw hash is never returned to the engine, and its values are never interpolated except through :func:~hyera.backends._eyaml.decrypt_string's own trailing context. interpolate call.

Source code in src/hyera/backends/_base.py

def __init__(
    self,
    conf: _ty.Optional[_ty.Mapping[str, _ty.Any]] = None,
    *,
    strict: _ty.Optional[_ty.Union[Strict, str]] = None,
) -> None:
    if conf is not None and not isinstance(conf, _abc.Mapping):
        raise TypeError(
            "conf must be a mapping or None, not {}".format(type(conf).__name__)
        )
    self.conf: _ty.Mapping[str, _ty.Any] = conf or {}
    if strict is not None and strict not in _STRICT_VALUES:
        raise ValueError(
            "strict must be one of {!r}, not {!r}".format(_STRICT_VALUES, strict)
        )
    self._strict = _plain(strict)
    self.name: _ty.Optional[str] = type(self)._default_name()

NAMES = {'function': ('eyaml_lookup_key',)} class-attribute

check_available() classmethod

Raise :class:BackendError naming the eyaml extra when cryptography is not importable.

Source code in src/hyera/backends/_eyaml.py

@classmethod
def check_available(cls) -> None:
    """Raise :class:`BackendError` naming the ``eyaml`` extra
    when ``cryptography`` is not importable."""
    check_cryptography()

lookup_key(key, options, context)

Decrypt key's PKCS7 ENC[...] value from the .eyaml file named by the hierarchy location, matching Puppet's eyaml_lookup_key.

Parameters:

Name	Type	Description	Default
key	str	the key to decrypt.	required
options	Mapping[str, Any]	the hierarchy entry's options (path required).	required
context	LookupContext	the per-location :class:LookupContext.	required

Returns:

Type	Description
Any	the decrypted (and interpolated) value.

Raises:

Type	Description
ConfigError	no path location was declared.
BackendError	the private key or ciphertext could not be read, parsed or decrypted.

Source code in src/hyera/backends/_eyaml.py

def lookup_key(
    self,
    key: str,
    options: _ty.Mapping[str, _ty.Any],
    context: LookupContext,
) -> _ty.Any:
    """Decrypt ``key``'s PKCS7 ``ENC[...]`` value from the ``.eyaml``
    file named by the hierarchy location, matching Puppet's
    ``eyaml_lookup_key``.

    :param key: the key to decrypt.
    :param options: the hierarchy entry's ``options`` (``path`` required).
    :param context: the per-location :class:`LookupContext`.
    :returns: the decrypted (and interpolated) value.
    :raises ConfigError: no ``path`` location was declared.
    :raises BackendError: the private key or ciphertext could not be
        read, parsed or decrypted.
    """
    if "path" not in options:
        raise ConfigError(
            "'eyaml_lookup_key': one of 'path', 'paths' 'glob', 'globs' "
            "or 'mapped_paths' must be declared in hiera.yaml when "
            "using this lookup_key function"
        )
    path = options["path"]
    parsed = context.cached_file_data(path, parse=YAMLBackend().loads)

    # The non-Hash rule reads `self.strict` at call time, same as
    # `yaml_data`'s own -- applied fresh on every read (never cached),
    # so a later call under different strictness sees its own rule.
    raw = YAMLBackend(strict=self.strict)._as_data_hash(parsed, path)
    if key not in raw:
        context.not_found()
    value = raw[key]
    return self._decrypt(value, options, context, key, path)

HOCONBackend(conf=None, *, strict=None, hocon_includes=None, hocon_env=None)

Bases: Backend

HOCON (.conf) data via the optional pyhocon package.

Always registered: a missing/broken pyhocon fails at :meth:check_available (backend/level construction) and again in :meth:loads, both naming the hocon extra -- so the failure is always reachable instead of silently disappearing from :func:default_backends.

include directives resolve exactly as Puppet's own hocon_data does by default (see :func:_allow_hocon_includes for the rule): a plain quoted include contributes nothing, include file(...) really reads the file (relative to the process cwd, or absolute), and every other form (url(...), classpath(...), required(...), package(...), a case-mismatched keyword, a bare include with nothing valid after it) raises, matching Puppet's own parse/method errors for those forms. A value-position directive (including inside a [...] array) is kept as literal text, as Puppet keeps it.

Passing hocon_includes=False (to the constructor directly, or via a hocon_includes: false key on the hierarchy entry/defaults -- hyera's own extension, not Puppet vocabulary) restricts it instead (see :func:_refuse_hocon_includes): every directive form other than a plain quoted include raises, including file(...).

In either mode, pyhocon's own include entry points are also wrapped (see :func:_install_hocon_include_guard) as a fail-closed backstop for the forms that mode does not intend to resolve for real, so a gap in the text scanner still fails closed instead of silently reading a file or reaching the network. Durations are parsed via a private module copy (see :func:_hocon_parser) so they stay text.

Parameters:

Name	Type	Description	Default
conf	Optional[Mapping[str, Any]]	the hierarchy entry's/defaults's own mapping.	None
strict	Optional[str]	overrides the call-time default.	None
hocon_includes	Optional[bool]	None (the default) reads conf.get("hocon_includes", True).	None
hocon_env	Optional[bool]	None (the default) reads conf.get("hocon_env", True). When false, a substitution the document does not define is never taken from the process environment: ${?VAR} is absent and ${VAR} fails to resolve. Hyera's own extension.	None

Source code in src/hyera/backends/_hocon.py

def __init__(
    self,
    conf: _ty.Optional[_ty.Mapping[str, _ty.Any]] = None,
    *,
    strict: _ty.Optional[str] = None,
    hocon_includes: _ty.Optional[bool] = None,
    hocon_env: _ty.Optional[bool] = None,
) -> None:
    super().__init__(conf, strict=strict)
    if hocon_env is None:
        hocon_env = self.conf.get("hocon_env", True)
    self.hocon_env: bool = bool(hocon_env)
    # No `Hiera(backend_options=...)` plumbing exists, so the opt-in reads from the
    # level's own `conf` (its hiera.yaml hierarchy-entry/`defaults` mapping) when
    # not passed directly.
    if hocon_includes is None:
        hocon_includes = self.conf.get("hocon_includes", True)
    self.hocon_includes: bool = bool(hocon_includes)

EXTENSIONS = ('.conf',) class-attribute

NAMES = {'function': ('hocon_data',), 'format': ('hocon',)} class-attribute

hocon_env = bool(hocon_env) instance-attribute

hocon_includes = bool(hocon_includes) instance-attribute

check_available() classmethod

Raise :class:BackendError naming the hocon extra when pyhocon is not importable.

Source code in src/hyera/backends/_hocon.py

@classmethod
def check_available(cls) -> None:
    """Raise :class:`BackendError` naming the ``hocon`` extra
    when ``pyhocon`` is not importable."""
    if not has_hocon():
        raise BackendError(cls._MISSING_DEP_MESSAGE)

data_hash(path, options, context)

The data_hash hook. Besides path, the hierarchy options accepted are hocon_includes (a Boolean), which selects the include mode for this level, and hocon_env (a Boolean), which allows or stops reading the process environment; any other option raises as for every built-in file function.

Parameters:

Name	Type	Description	Default
path	Any	the location's file path.	required
options	Mapping[str, Any]	the hierarchy entry's options.	required
context	LookupContext	the per-location :class:LookupContext.	required

Returns:

Type	Description
Dict[str, Any]	the parsed data.

Raises:

Type	Description
ConfigError	hocon_includes or hocon_env is not a Boolean, or options carries anything else besides path.
BackendError	the file could not be read or parsed.

Source code in src/hyera/backends/_hocon.py

def data_hash(
    self,
    path: _ty.Any,
    options: _ty.Mapping[str, _ty.Any],
    context: LookupContext,
) -> _ty.Dict[str, _ty.Any]:
    """The ``data_hash`` hook. Besides ``path``, the hierarchy options
    accepted are ``hocon_includes`` (a Boolean), which selects the include
    mode for this level, and ``hocon_env`` (a Boolean), which allows or
    stops reading the process environment; any other option raises as
    for every built-in file function.

    :param path: the location's file path.
    :param options: the hierarchy entry's ``options``.
    :param context: the per-location :class:`LookupContext`.
    :returns: the parsed data.
    :raises ConfigError: ``hocon_includes`` or ``hocon_env`` is not a Boolean, or
        ``options`` carries anything else besides ``path``.
    :raises BackendError: the file could not be read or parsed.
    """
    rest = dict(options)
    chosen = {}
    for name in ("hocon_includes", "hocon_env"):
        if name not in rest:
            continue
        value = rest.pop(name)
        if not isinstance(value, bool):
            raise ConfigError(
                "'hocon_data' option '{}' must be a Boolean, not {}".format(
                    name, type(value).__name__
                )
            )
        chosen[name] = value
    backend = self
    if any(getattr(self, name) != value for name, value in chosen.items()):
        settings = {
            "hocon_includes": self.hocon_includes,
            "hocon_env": self.hocon_env,
        }
        settings.update(chosen)
        backend = type(self)(self.conf, strict=self._strict, **settings)
    return super(HOCONBackend, backend).data_hash(path, rest, context)

loads(text)

Parse HOCON the way Puppet's hocon_data does: include file(...) really reads the file, include url(...)/ classpath(...)/required(...) and durations raise/stay text (see the class docstring for the full fidelity rule).

Parameters:

Name	Type	Description	Default
text	str	the HOCON text to parse.	required

Returns:

Type	Description
Any	the parsed value.

Raises:

Type	Description
BackendError	pyhocon is missing, or text is not valid HOCON (or an include/duration form this rule rejects).

Source code in src/hyera/backends/_hocon.py

def loads(self, text: str) -> _ty.Any:
    """Parse HOCON the way Puppet's ``hocon_data`` does: ``include
    file(...)`` really reads the file, ``include url(...)``/
    ``classpath(...)``/``required(...)`` and durations raise/stay text
    (see the class docstring for the full fidelity rule).

    :param text: the HOCON text to parse.
    :returns: the parsed value.
    :raises BackendError: ``pyhocon`` is missing, or ``text`` is not
        valid HOCON (or an ``include``/duration form this rule rejects).
    """
    try:
        from pyhocon import ConfigTree
    except ImportError:
        raise BackendError(self._MISSING_DEP_MESSAGE) from None
    except Exception as e:
        raise BackendError(
            "hocon_data backend could not import 'pyhocon' ({}: {}); "
            'pip install "hyera[hocon]"'.format(type(e).__name__, e)
        ) from None
    if self.hocon_includes:
        text = _allow_hocon_includes(text)
        # `file(...)` is deliberately NOT guarded here: running it for real matches
        # Puppet. `url`/`package` stay backstopped, as Puppet's hocon_data cannot
        # resolve those either.
        guarded = frozenset({"URL include", "package include"})
    else:
        text = _refuse_hocon_includes(text)
        guarded = frozenset({"file include", "URL include", "package include"})
    token = _HOCON_INCLUDE_GUARD.set(guarded)
    env_token = _HOCON_ENV.set(self.hocon_env)
    failure = None
    try:
        mod = _hocon_parser()
        parsed = mod.ConfigFactory.parse_string(text)
    except BackendError:
        raise
    except Exception as e:
        failure = _one_line(str(e))
    finally:
        _HOCON_ENV.reset(env_token)
        _HOCON_INCLUDE_GUARD.reset(token)
    if failure is not None:
        # Raised outside the handler: pyhocon's exception carries the
        # whole document and must not stay reachable from this one.
        raise BackendError(failure) from None
    if not isinstance(parsed, ConfigTree):
        raise BackendError(
            "hocon_data: has type {} rather than object at file "
            "root".format(_hocon_root_kind(parsed))
        )
    return _as_plain(parsed)

JSONBackend(conf=None, *, strict=None)

Bases: Backend

JSON (.json) data, matching Ruby's json gem: /* *// // ... comments allowed, NaN/Infinity/-Infinity and a lone (unpaired) surrogate code point rejected.

Source code in src/hyera/backends/_base.py

def __init__(
    self,
    conf: _ty.Optional[_ty.Mapping[str, _ty.Any]] = None,
    *,
    strict: _ty.Optional[_ty.Union[Strict, str]] = None,
) -> None:
    if conf is not None and not isinstance(conf, _abc.Mapping):
        raise TypeError(
            "conf must be a mapping or None, not {}".format(type(conf).__name__)
        )
    self.conf: _ty.Mapping[str, _ty.Any] = conf or {}
    if strict is not None and strict not in _STRICT_VALUES:
        raise ValueError(
            "strict must be one of {!r}, not {!r}".format(_STRICT_VALUES, strict)
        )
    self._strict = _plain(strict)
    self.name: _ty.Optional[str] = type(self)._default_name()

EXTENSIONS = ('.json',) class-attribute

NAMES = {'function': ('json_data',), 'format': ('json',)} class-attribute

dumps(obj, **kw)

Render obj as JSON, with non-ASCII characters left as-is.

Parameters:

Name	Type	Description	Default
obj	Any	the value to render.	required
kw	Any	forwarded to :func:json.dumps.	{}

Returns:

Type	Description
str	the rendered JSON text.

Source code in src/hyera/backends/_json.py

def dumps(self, obj: _ty.Any, **kw: _ty.Any) -> str:
    """Render ``obj`` as JSON, with non-ASCII characters left as-is.

    :param obj: the value to render.
    :param kw: forwarded to :func:`json.dumps`.
    :returns: the rendered JSON text.
    """
    kw.setdefault("ensure_ascii", False)
    return json.dumps(obj, **kw)

loads(text)

Parse JSON the way Ruby's json gem does: /* *//// comments allowed, NaN/Infinity/-Infinity and a lone surrogate rejected.

Parameters:

Name	Type	Description	Default
text	str	the JSON text to parse.	required

Returns:

Type	Description
Any	the parsed value.

Raises:

Type	Description
BackendError	text is not valid JSON by that rule.

Source code in src/hyera/backends/_json.py

def loads(self, text: str) -> _ty.Any:
    """Parse JSON the way Ruby's ``json`` gem does: ``/* */``/``//``
    comments allowed, ``NaN``/``Infinity``/``-Infinity`` and a lone
    surrogate rejected.

    :param text: the JSON text to parse.
    :returns: the parsed value.
    :raises BackendError: ``text`` is not valid JSON by that rule.
    """
    problem = None
    try:
        result = loads_json(_strip_json_comments(text))
        check_json_value(result)
        return result
    except json.JSONDecodeError as e:
        problem = "{} at line {} column {}".format(e.msg, e.lineno, e.colno)
    except RecursionError:
        problem = TOO_DEEP
    except ValueError as e:
        problem = str(e)
    # Outside the except block, matching YAMLBackend's chain-free style.
    raise BackendError(problem)

LookupContext(function_context, invocation)

The context argument handed to a lookup_key/data_dig backend hook (Puppet's public Context API, context.rb:126-206).

Built by the engine for each call; a backend never constructs one itself. A test builds one with :meth:for_testing.

Parameters:

Name	Type	Description	Default
function_context	_FunctionContext	the per-location state to read/write through.	required
invocation	Invocation	the current lookup's per-lookup state.	required

Source code in src/hyera/_lookup/function_provider.py

def __init__(
    self, function_context: _FunctionContext, invocation: Invocation
) -> None:
    self._fc = function_context
    self._invocation = invocation
    #: ``(path, stamp)`` of every file read through
    #: :meth:`cached_file_data` by this call.
    self._deps: _ty.List[_ty.Tuple[str, _ty.Any]] = []

environment_name property

The current lookup's environment name, or None.

module_name property

The current hierarchy entry's module name, or None at the global/environment layer.

cache(key, value)

Cache value under key for this location, for the life of the owning Hiera/h.scoped(...) view. Returns value.

Parameters:

Name	Type	Description	Default
key	Any	the cache key.	required
value	Any	the value to cache.	required

Returns:

Type	Description
Any	value, unchanged.

Source code in src/hyera/_lookup/function_provider.py

def cache(self, key: _ty.Any, value: _ty.Any) -> _ty.Any:
    """Cache ``value`` under ``key`` for this location, for the life of
    the owning ``Hiera``/``h.scoped(...)`` view. Returns ``value``.

    :param key: the cache key.
    :param value: the value to cache.
    :returns: ``value``, unchanged.
    """
    self._fc._cache[key] = value
    return value

cache_all(mapping)

:meth:cache every key/value pair of mapping.

Parameters:

Name	Type	Description	Default
mapping	Mapping[Any, Any]	the key/value pairs to cache.	required

Source code in src/hyera/_lookup/function_provider.py

def cache_all(self, mapping: _ty.Mapping[_ty.Any, _ty.Any]) -> None:
    """:meth:`cache` every key/value pair of ``mapping``.

    :param mapping: the key/value pairs to cache.
    """
    self._fc._cache.update(mapping)

cache_has_key(key)

Whether key was already :meth:cache\ d for this location.

Parameters:

Name	Type	Description	Default
key	Any	the cache key.	required

Returns:

Type	Description
bool	whether key is cached.

Source code in src/hyera/_lookup/function_provider.py

def cache_has_key(self, key: _ty.Any) -> bool:
    """Whether ``key`` was already :meth:`cache`\\ d for this location.

    :param key: the cache key.
    :returns: whether ``key`` is cached.
    """
    return key in self._fc._cache

cached_entries()

An iterator over every (key, value) pair :meth:cache\ d for this location.

Returns:

Type	Description
'_ty.Iterator[_ty.Tuple[_ty.Any, _ty.Any]]'	an iterator of (key, value) pairs.

Source code in src/hyera/_lookup/function_provider.py

def cached_entries(self) -> "_ty.Iterator[_ty.Tuple[_ty.Any, _ty.Any]]":
    """An iterator over every ``(key, value)`` pair :meth:`cache`\\ d
    for this location.

    :returns: an iterator of ``(key, value)`` pairs.
    """
    return iter(list(self._fc._cache.items()))

cached_file_data(path, parse=None)

The cached result of parse(text) (or the raw text when parse is None) for the file at path, shared with every other location of this Hiera instance (see :class:_EnvironmentContext).

Parameters:

Name	Type	Description	Default
path	Union[str, 'os.PathLike[str]']	the file to read.	required
parse	Optional[Callable[[str], Any]]	applied to the file's text; identity when omitted.	None

Returns:

Type	Description
Any	the (cached) parsed result, or raw text.

Raises:

Type	Description
BackendError	path could not be read, decoded as UTF-8, or parse raised a :class:BackendError of its own.

Source code in src/hyera/_lookup/function_provider.py

def cached_file_data(
    self,
    path: _ty.Union[str, "os.PathLike[str]"],
    parse: _ty.Optional[_ty.Callable[[str], _ty.Any]] = None,
) -> _ty.Any:
    """The cached result of ``parse(text)`` (or the raw text when
    ``parse`` is ``None``) for the file at ``path``, shared with every
    other location of this ``Hiera`` instance (see
    :class:`_EnvironmentContext`).

    :param path: the file to read.
    :param parse: applied to the file's text; identity when omitted.
    :returns: the (cached) parsed result, or raw text.
    :raises BackendError: ``path`` could not be read, decoded as UTF-8,
        or ``parse`` raised a :class:`BackendError` of its own.
    """
    stamp, value = self._fc.environment_context.cached_file_stamped(path, parse)
    self._deps.append((os.fspath(path), stamp))
    return value

cached_value(key)

The value :meth:cache\ d under key, or None.

Parameters:

Name	Type	Description	Default
key	Any	the cache key.	required

Returns:

Type	Description
Any	the cached value, or None.

Source code in src/hyera/_lookup/function_provider.py

def cached_value(self, key: _ty.Any) -> _ty.Any:
    """The value :meth:`cache`\\ d under ``key``, or ``None``.

    :param key: the cache key.
    :returns: the cached value, or ``None``.
    """
    return self._fc._cache.get(key)

explain(producer)

Add producer's text to this lookup's explain() report.

Parameters:

Name	Type	Description	Default
producer	Callable[[], str]	a zero-argument callable producing the text.	required

Source code in src/hyera/_lookup/function_provider.py

def explain(self, producer: _ty.Callable[[], str]) -> None:
    """Add ``producer``'s text to this lookup's ``explain()`` report.

    :param producer: a zero-argument callable producing the text.
    """
    self._invocation.report_text(producer)

for_testing(*, scope=None, module_name=None, data=None) classmethod

A context to call a hook with outside a lookup.

:meth:cache, :meth:cached_file_data and :meth:explain behave as in a lookup, except that :meth:explain never calls its producer.

Parameters:

Name	Type	Description	Default
scope	Optional[Scope]	what :meth:interpolate reads variables from, and the source of :attr:environment_name; an empty :class:~hyera.Scope when omitted.	None
module_name	Optional[str]	the value of :attr:module_name.	None
data	Optional[Mapping[str, Any]]	the keys lookup(), alias() and hiera() resolve in :meth:interpolate, by dotted navigation; a key absent from it is a miss, as in a lookup. No keys when omitted.	None

Returns:

Type	Description
'LookupContext'	the new context.

Source code in src/hyera/_lookup/function_provider.py

@classmethod
def for_testing(
    cls,
    *,
    scope: _ty.Optional[Scope] = None,
    module_name: _ty.Optional[str] = None,
    data: _ty.Optional[_ty.Mapping[str, _ty.Any]] = None,
) -> "LookupContext":
    """A context to call a hook with outside a lookup.

    :meth:`cache`, :meth:`cached_file_data` and :meth:`explain` behave as
    in a lookup, except that :meth:`explain` never calls its producer.

    :param scope: what :meth:`interpolate` reads variables from, and the
        source of :attr:`environment_name`; an empty :class:`~hyera.Scope`
        when omitted.
    :param module_name: the value of :attr:`module_name`.
    :param data: the keys ``lookup()``, ``alias()`` and ``hiera()`` resolve
        in :meth:`interpolate`, by dotted navigation; a key absent from it
        is a miss, as in a lookup. No keys when omitted.
    :returns: the new context.
    """
    scope = Scope() if scope is None else scope
    keys = {} if data is None else data

    def lookup(key, invocation):
        root, segments = parse_lookup_key(key)
        if root not in keys:
            return _MISSING
        if not segments:
            return keys[root]
        return sub_lookup(key, segments, keys[root])

    function_context = _FunctionContext(
        _EnvironmentContext(), scope.environment, module_name
    )
    return cls(function_context, Invocation(scope, lookup))

interpolate(value)

Interpolate value (methods allowed) against the current lookup's scope -- a backend calls this itself; the engine never interpolates a lookup_key/data_dig result on its own.

Parameters:

Name	Type	Description	Default
value	Any	the value (or nested structure) to interpolate.	required

Returns:

Type	Description
Any	the interpolated result.

Source code in src/hyera/_lookup/function_provider.py

def interpolate(self, value: _ty.Any) -> _ty.Any:
    """Interpolate ``value`` (methods allowed) against the current
    lookup's scope -- a backend calls this itself; the engine never
    interpolates a ``lookup_key``/``data_dig`` result on its own.

    :param value: the value (or nested structure) to interpolate.
    :returns: the interpolated result.
    """
    return interpolate(value, self._invocation, allow_methods=True)

not_found()

Signal a miss for this location -- Puppet's throw :no_such_key.

Raises:

Type	Description
Exception	always; the raised object is an internal control-flow signal, not a documented public exception.

Source code in src/hyera/_lookup/function_provider.py

def not_found(self) -> "_ty.NoReturn":
    """Signal a miss for this location -- Puppet's ``throw :no_such_key``.

    :raises Exception: always; the raised object is an internal
        control-flow signal, not a documented public exception.
    """
    raise _NotFound()

NamePattern

Bases: NamedTuple

A registered name that matches by regex instead of exact string.

regex is matched with :meth:re.Pattern.fullmatch; its named groups are passed as keyword arguments to the backend's constructor.

display instance-attribute

regex instance-attribute

RubySymbol(name)

A Ruby :symbol value (!ruby/sym/!ruby/symbol, or a plain :name scalar). Not a str subclass -- no string-typed code path should ever accept one by accident.

Parameters:

Name	Type	Description	Default
name	str	the symbol's name, without the leading :.	required

Source code in src/hyera/backends/_psych.py

def __init__(self, name: str) -> None:
    self._name: str = name

__slots__ = ('_name',) class-attribute instance-attribute

name property

The symbol's name, without the leading :.

__eq__(other)

Source code in src/hyera/backends/_psych.py

def __eq__(self, other: object) -> bool:
    if not isinstance(other, RubySymbol):
        return NotImplemented
    return self._name == other._name

__hash__()

Source code in src/hyera/backends/_psych.py

def __hash__(self) -> int:
    return hash((RubySymbol, self._name))

__ne__(other)

Source code in src/hyera/backends/_psych.py

def __ne__(self, other: object) -> bool:
    if not isinstance(other, RubySymbol):
        return NotImplemented
    return self._name != other._name

__reduce__()

Source code in src/hyera/backends/_psych.py

def __reduce__(self) -> "tuple[type, tuple[str]]":
    return (RubySymbol, (self._name,))

__repr__()

Source code in src/hyera/backends/_psych.py

def __repr__(self) -> str:
    return ":{}".format(self.name)

SopsBackend(conf=None, *, strict=None, format=None, timeout=None)

Bases: Backend

Decrypted on the fly via the sops CLI (sops_data; the one kept non-Puppet deviation).

Hardened for unattended use: the subprocess has a finite timeout and its whole process group is killed when it expires (a :class:BackendTimeoutError), its stdin is the null device, its stderr is captured and surfaced (the last 2,000 characters), a missing sops binary raises a clear :class:BackendError, and a decrypted file that fails to parse reports only the problem and its line/column -- never the decrypted plaintext.

sops_data and sops infer the format from the file extension with sops's own rule (:data:_SOPS_SUFFIXES); the sops_<format> name (a :class:NamePattern, added in a later commit alongside the sops alias) forces one regardless of extension via the format constructor keyword.

Parameters:

Name	Type	Description	Default
conf	Optional[Mapping[str, Any]]	the hierarchy entry's/defaults's own mapping.	None
strict	Optional[str]	overrides the call-time default.	None
format	Optional[str]	forces the decrypted plaintext's format (yaml/json/ini/dotenv) regardless of extension.	None
timeout	Optional[float]	seconds to wait for sops; None reads hyera.backends.SOPS_TIMEOUT at each call.	None

Raises:

Type	Description
ConfigError	timeout is not a positive number.

Source code in src/hyera/backends/_sops.py

def __init__(
    self,
    conf: _ty.Optional[_ty.Mapping[str, _ty.Any]] = None,
    *,
    strict: _ty.Optional[str] = None,
    format: _ty.Optional[str] = None,
    timeout: _ty.Optional[float] = None,
) -> None:
    super().__init__(conf, strict=strict)
    self._format = format
    if timeout is not None and (
        isinstance(timeout, bool)
        or not isinstance(timeout, (int, float))
        or not timeout > 0
    ):
        raise ConfigError(
            "sops timeout must be a positive number of seconds, not "
            "{!r}".format(timeout)
        )
    self._timeout = timeout

NAMES = {'function': ('sops_data', 'sops', NamePattern('sops_<yaml|json|ini|dotenv>', re.compile('sops_(?P<format>yaml|json|ini|dotenv)')))} class-attribute

data_hash(path, options, context)

Decrypt path with the sops CLI and parse the plaintext in the format sops itself reports for it (or the format constructor keyword, when given).

Parameters:

Name	Type	Description	Default
path	'Path'	the encrypted file's location.	required
options	Mapping[str, Any]	the hierarchy entry's options.	required
context	LookupContext	the per-location :class:LookupContext.	required

Returns:

Type	Description
Dict[str, Any]	the decrypted, parsed data.

Raises:

Type	Description
ConfigError	path has no recognized suffix and no format was given, or options carries anything besides path.
BackendError	sops is missing, times out, exits non-zero, or the decrypted plaintext could not be parsed.

Source code in src/hyera/backends/_sops.py

def data_hash(
    self,
    path: "Path",
    options: _ty.Mapping[str, _ty.Any],
    context: LookupContext,
) -> _ty.Dict[str, _ty.Any]:
    """Decrypt ``path`` with the ``sops`` CLI and parse the plaintext
    in the format sops itself reports for it (or the ``format``
    constructor keyword, when given).

    :param path: the encrypted file's location.
    :param options: the hierarchy entry's ``options``.
    :param context: the per-location :class:`LookupContext`.
    :returns: the decrypted, parsed data.
    :raises ConfigError: ``path`` has no recognized suffix and no
        ``format`` was given, or ``options`` carries anything besides
        ``path``.
    :raises BackendError: ``sops`` is missing, times out, exits
        non-zero, or the decrypted plaintext could not be parsed.
    """
    self._require_path_only(path, options)
    fmt = self._format or _sops_format(str(path))
    if fmt is None:
        raise ConfigError(
            "sops_data: '{}' has no .yaml/.yml/.json/.env/.ini suffix, "
            "so sops reads it as binary, which is not a data hash; use "
            "data_hash: sops_<yaml|json|ini|dotenv> to choose a format"
            "".format(path)
        )
    # sops's INI *writer* is ambiguous: a decrypted value with `"""` and a newline
    # can inject a key or section undetectably. INI is decrypted as sops's JSON view
    # (`{"DEFAULT": {...}, section: {...}}`) and parsed with JSONBackend.
    parse_fmt = "json" if fmt == "ini" else fmt
    raw = _run_sops(path, fmt, output_type=parse_fmt, timeout=self._timeout)
    format_backend = Backend.new(parse_fmt, kind="format", strict=self.strict)
    problem = None
    text = None
    try:
        text = raw.decode("utf-8")
        parsed = format_backend.loads(text)
    except UnicodeDecodeError as e:
        # Optional hardening: the byte offset only, never the
        # offending byte value or the surrounding text the stock
        # codec message quotes.
        problem = "invalid UTF-8 at byte offset {}".format(e.start)
    except BackendError as e:
        problem = _sops_redact_yaml_problem(str(e))
    else:
        return format_backend._as_data_hash(parsed, path)
    finally:
        # Optional hardening: drop the plaintext locals before the
        # raise below, so a frame-capturing error reporter (e.g.
        # Sentry's default) does not also collect them.
        del raw, text
    raise BackendError(
        "Unable to parse ({}): {}".format(path, problem), path=str(path)
    )

YAMLBackend(conf=None, *, strict=None)

Bases: Backend

YAML (.yaml/.yml) data via Puppet's own Psych-compatible rules (:mod:hyera.backends._psych): numbers/booleans/dates/symbols parse Ruby's way, and a non-Hash top-level document warns (or raises under strict="error") and reads as empty.

Source code in src/hyera/backends/_base.py

def __init__(
    self,
    conf: _ty.Optional[_ty.Mapping[str, _ty.Any]] = None,
    *,
    strict: _ty.Optional[_ty.Union[Strict, str]] = None,
) -> None:
    if conf is not None and not isinstance(conf, _abc.Mapping):
        raise TypeError(
            "conf must be a mapping or None, not {}".format(type(conf).__name__)
        )
    self.conf: _ty.Mapping[str, _ty.Any] = conf or {}
    if strict is not None and strict not in _STRICT_VALUES:
        raise ValueError(
            "strict must be one of {!r}, not {!r}".format(_STRICT_VALUES, strict)
        )
    self._strict = _plain(strict)
    self.name: _ty.Optional[str] = type(self)._default_name()

EXTENSIONS = ('.yaml', '.yml') class-attribute

NAMES = {'function': ('yaml_data',), 'format': ('yaml',)} class-attribute

dumps(obj, **kw)

Render obj as YAML (block style, sorted keys off, Unicode left unescaped).

Parameters:

Name	Type	Description	Default
obj	Any	the value to render.	required
kw	Any	forwarded to :func:yaml.safe_dump.	{}

Returns:

Type	Description
str	the rendered YAML text.

Source code in src/hyera/backends/_yaml.py

def dumps(self, obj: _ty.Any, **kw: _ty.Any) -> str:
    """Render ``obj`` as YAML (block style, sorted keys off, Unicode
    left unescaped).

    :param obj: the value to render.
    :param kw: forwarded to :func:`yaml.safe_dump`.
    :returns: the rendered YAML text.
    """
    kw.setdefault("sort_keys", False)
    kw.setdefault("allow_unicode", True)
    kw.setdefault("default_flow_style", False)
    return yaml.safe_dump(obj, **kw)

loads(text)

Parse YAML the way Puppet's yaml_data does (Ruby Psych semantics via :mod:hyera.backends._psych), not PyYAML's own Python-flavored resolver.

Parameters:

Name	Type	Description	Default
text	str	the YAML text to parse.	required

Returns:

Type	Description
Any	the parsed value.

Source code in src/hyera/backends/_yaml.py

def loads(self, text: str) -> _ty.Any:
    """Parse YAML the way Puppet's ``yaml_data`` does (Ruby Psych
    semantics via :mod:`hyera.backends._psych`), not PyYAML's own
    Python-flavored resolver.

    :param text: the YAML text to parse.
    :returns: the parsed value.
    """
    # Psych's rules (types, BOM, one-document, symbol keys/values),
    # ported in ``_psych``: numbers/booleans/dates/symbols per
    # Ruby's ScalarScanner, not PyYAML's own Python-flavored resolver.
    limits = self.limits
    return safe_load(text, limits.yaml_alias_nodes if limits is not None else None)

default_backends()

The distinct backend classes registered in the function namespace, in definition order (YAML, JSON, HOCON, sops, eyaml).

Returns:

Type	Description
'_ty.List[_ty.Type[Backend]]'	the default Hiera(backends=...) allow-list.

Source code in src/hyera/backends/_base.py

def default_backends() -> "_ty.List[_ty.Type[Backend]]":
    """The distinct backend classes registered in the ``function``
    namespace, in definition order (YAML, JSON, HOCON, sops, eyaml).

    :returns: the default ``Hiera(backends=...)`` allow-list.
    """
    _load_entry_points()
    registry = Backend._REGISTRY.get("function", {"exact": {}, "patterns": []})
    seen = []
    for cls in registry["exact"].values():
        if cls not in seen:
            seen.append(cls)
    for _pattern, cls in registry["patterns"]:
        if cls not in seen:
            seen.append(cls)
    return seen

has_hocon()

True iff the optional pyhocon dependency imports without error.

Any import-time exception (not just ImportError -- an installed but broken pyhocon can raise something else entirely, e.g. AttributeError against a too-new stdlib) is caught and logged at debug, so a broken optional dependency never breaks every Hiera().

Returns:

Type	Description
bool	whether pyhocon is usable.

Source code in src/hyera/backends/_hocon.py

def has_hocon() -> bool:
    """True iff the optional ``pyhocon`` dependency imports without error.

    Any import-time exception (not just ``ImportError`` -- an installed but
    broken ``pyhocon`` can raise something else entirely, e.g.
    ``AttributeError`` against a too-new stdlib) is caught and logged at
    debug, so a broken optional dependency never breaks every ``Hiera()``.

    :returns: whether ``pyhocon`` is usable.
    """
    try:
        import pyhocon  # noqa: F401

        return True
    except Exception as e:
        _LOGGER.debug("pyhocon is not usable: %s: %s", type(e).__name__, e)
        return False