Sources

anycfg.sources: what a loader reads that is not a file path, and the registry of <scheme>:// strings. See the Sources guide for how each kind of source is resolved.

The contract

anycfg.sources.Source

Base class for something a loader reads that is not a file path.

Override :meth:read. Set executes = True for a source that runs a program: the loader then refuses it unless the policy allows commands, before :meth:read is called.

executes = False class-attribute

name property

A name for messages and for the default merge key.

read(context)

Return what this source holds, under context's policy.

Source code in src/anycfg/sources/_base.py

def read(self, context: "LoadContext") -> Payload:
    """Return what this source holds, under *context*'s policy."""
    raise NotImplementedError

anycfg.sources.Payload

Bases: NamedTuple

What a source read.

Attributes:

Name	Type	Description
data	Any	Text or bytes for a format to parse, or the data itself when parsed is true.
name	str	The source's name, for messages and merge keys.
identity	str	A stable key for this source, used for include-cycle detection and de-duplication.
format	'_ty.Optional[str]'	The format name the source itself determined (cmd+json://, a #!json line), or None to let the caller decide.
origin	Any	The location relative includes resolve against, or None.
parsed	bool	data is already data, not text.
trusted_format	bool	format was chosen by the caller, so it may name any registered format; one a source read from data (a command's #!json line) is limited to the formats documents may select.
encoding	'_ty.Optional[str]'	The codec data is in, and the default for its includes; None means the load's.

data instance-attribute

encoding = None class-attribute instance-attribute

format = None class-attribute instance-attribute

identity instance-attribute

name instance-attribute

origin = None class-attribute instance-attribute

parsed = False class-attribute instance-attribute

trusted_format = False class-attribute instance-attribute

Built-in sources

anycfg.sources.Text(data, *, format=None, name='')

Bases: Source

In-memory content in a named format: Text("a: 1", format="yaml").

data is str or bytes (decoded with the load's encoding). Without format, the content must name its own: a #!<format> first line (which is removed) or a format's marker such as YAML's %YAML directive; such a format is limited to the ones documents may select. Relative includes inside it resolve against the loader's base_dir. name appears in messages and is the merge key for collect="hash".

Source code in src/anycfg/sources/_base.py

def __init__(
    self,
    data: "_ty.Union[str, bytes]",
    *,
    format: "_ty.Optional[str]" = None,
    name: str = "",
) -> None:
    if not isinstance(data, (str, bytes, bytearray)):
        raise TypeError(f"Text takes str or bytes, not {type(data).__name__}")
    self.data = bytes(data) if isinstance(data, bytearray) else data
    self.format = format
    self._name = name

data = bytes(data) if isinstance(data, bytearray) else data instance-attribute

format = format instance-attribute

name property

__repr__()

Source code in src/anycfg/sources/_base.py

def __repr__(self) -> str:
    return f"Text(format={self.format!r}, name={self._name!r})"

read(context)

Source code in src/anycfg/sources/_base.py

def read(self, context: "LoadContext") -> Payload:
    if self.format is not None:
        return Payload(
            self.data,
            name=self._name,
            identity=f"text:{id(self)}",
            format=self.format,
            trusted_format=True,
        )
    from .. import formats

    named, data = formats.declared(self.data)
    return Payload(data, name=self._name, identity=f"text:{id(self)}", format=named)

anycfg.sources.Chunks(chunks, *, format=None, name='')

Bases: Source

Content arriving in pieces: any iterable of str or bytes chunks.

Chunks(response.iter_content(), format="json"). The chunks are joined as they arrive and the read stops once they pass the policy's max_bytes. The format is format, else the one name's suffix selects. An iterator is read once, so a source built on one loads once.

Source code in src/anycfg/sources/_base.py

def __init__(
    self,
    chunks: _ty.Iterable[_ty.Any],
    *,
    format: "_ty.Optional[str]" = None,
    name: str = "",
) -> None:
    if isinstance(chunks, (str, bytes, bytearray)) or not isinstance(
        chunks, _ty.Iterable
    ):
        raise TypeError(
            f"Chunks takes an iterable of chunks, not {type(chunks).__name__}"
        )
    self.chunks = chunks
    self.format = format
    self._name = name

chunks = chunks instance-attribute

format = format instance-attribute

name property

__repr__()

Source code in src/anycfg/sources/_base.py

def __repr__(self) -> str:
    return f"Chunks(format={self.format!r}, name={self._name!r})"

read(context)

Source code in src/anycfg/sources/_base.py

def read(self, context: "LoadContext") -> Payload:
    data = _join_chunks(self.chunks, context.policy.max_bytes, self._name)
    return Payload(
        data,
        name=self._name,
        identity=f"chunks:{id(self)}",
        format=self.format,
        trusted_format=self.format is not None,
    )

anycfg.sources.Data(data, name='')

Bases: Source

Python data as a source: a mapping or a list, merged like a document.

The data is deep-copied when read, so a later merge or interpolation never reaches back into the caller's object. A Mapping or a list passed straight to a loader is wrapped in one of these.

Source code in src/anycfg/sources/_base.py

def __init__(self, data: _ty.Any, name: str = "") -> None:
    self.data = data
    self._name = name

data = data instance-attribute

name property

__repr__()

Source code in src/anycfg/sources/_base.py

def __repr__(self) -> str:
    return f"Data({self._name!r})"

read(context)

Source code in src/anycfg/sources/_base.py

def read(self, context: "LoadContext") -> Payload:
    return Payload(
        _copy_data(self.data, {}),
        name=self._name,
        identity=f"data:{id(self.data)}",
        parsed=True,
    )

anycfg.sources.Env(prefix='', *, nested=None, lowercase=False, coerce=False, environ=None)

Bases: Source

Environment variables as a configuration document.

Only variables starting with prefix are read, with the prefix removed. With nested ("__"), APP_DB__PORT becomes {"DB": {"PORT": ...}}; a variable that is both a value and a parent of nested keys raises, naming both. Values stay strings — types come from hydration (load(into=...)) — unless coerce is set: True reads each value by the YAML 1.2 core rules (80 an int, true a bool, null None; yes and 0755 stay text), a callable (text) -> value decides itself. Keys keep their case unless lowercase is set. On Windows the prefix is matched case-insensitively, as the environment is.

environ replaces os.environ, for tests and for a captured snapshot.

Source code in src/anycfg/sources/_env.py

def __init__(
    self,
    prefix: str = "",
    *,
    nested: "_ty.Optional[str]" = None,
    lowercase: bool = False,
    coerce: "Coerce" = False,
    environ: "_ty.Optional[_ty.Mapping[str, str]]" = None,
) -> None:
    self.prefix = prefix
    self.nested = nested
    self.lowercase = lowercase
    self.coerce = coerce
    self._coerce = coercer(coerce)
    self.environ = environ

coerce = coerce instance-attribute

environ = environ instance-attribute

lowercase = lowercase instance-attribute

name property

nested = nested instance-attribute

prefix = prefix instance-attribute

__repr__()

Source code in src/anycfg/sources/_env.py

def __repr__(self) -> str:
    return f"Env({self.prefix!r})"

read(context)

Source code in src/anycfg/sources/_env.py

def read(self, context: "LoadContext") -> Payload:
    environ = os.environ if self.environ is None else self.environ
    result: "_ty.Dict[str, _ty.Any]" = {}
    origins: "_ty.Dict[_ty.Tuple[str, ...], str]" = {}
    prefix = self.prefix
    for key, value in environ.items():
        if prefix and not _has_prefix(key, prefix):
            continue
        clean = key[len(prefix) :]
        if self._coerce is not None:
            value = self._coerce(value)
        if self.lowercase:
            clean = clean.lower()
        if not clean:
            # A variable equal to the prefix would produce a "" key.
            continue
        if self.nested and self.nested in clean:
            parts = [part for part in clean.split(self.nested) if part]
            if parts:
                _set_nested(result, parts, value, origins, key)
            continue
        if not self.nested:
            # Every variable is a leaf: no collision is possible, and this
            # is the common path.
            result[clean] = value
            continue
        if isinstance(result.get(clean), dict):
            raise _conflict(key, origins.get((clean,), "<unknown>"), (clean,))
        result[clean] = value
        origins[(clean,)] = key
    return Payload(result, name=self.name, identity=self.name, parsed=True)

anycfg.sources.Command(command, *, format=None, timeout=_UNSET, encoding=None)

Bases: Source

Run a command and parse its standard output.

command is either a scheme string — cmd://..., exec://... or sh://..., optionally naming the output format as cmd+json://... — or a bare command line. Both run through the shell. Use :meth:script for a script file, which runs through its interpreter without a shell.

The output format comes from, in order: format, the +fmt of the scheme, a #!<format> first line of the output (which is then dropped), and for a script the suffix before its own (secrets.json.sh); then the loader's own format=. Output with no format raises: content is never sniffed. Includes inside the output are read with the same encoding.

The command runs with stdin closed and a timeout (the policy's command_timeout unless timeout is given), after which it and its child processes are killed. Output is decoded strictly with encoding (the load's encoding by default). A non-zero exit raises :class:~anycfg.exceptions.CommandError, whose message carries the tail of stderr and never stdout.

Source code in src/anycfg/sources/_command.py

def __init__(
    self,
    command: str,
    *,
    format: "_ty.Optional[str]" = None,
    timeout: "_ty.Optional[float]" = _UNSET,
    encoding: "_ty.Optional[str]" = None,
) -> None:
    from . import SCHEME_REGEX

    self.text = command
    self.format = format
    self.timeout = timeout
    self.encoding = encoding
    self._script: "_ty.Optional[str]" = None
    match = SCHEME_REGEX.match(command)
    if match is not None:
        self.command = command[match.end() :]
        if self.format is None and match.group(2):
            self.format = match.group(2).lower()
    else:
        self.command = command

command = command[match.end():] instance-attribute

encoding = encoding instance-attribute

executes = True class-attribute instance-attribute

format = format instance-attribute

name property

text = command instance-attribute

timeout = timeout instance-attribute

__repr__()

Source code in src/anycfg/sources/_command.py

def __repr__(self) -> str:
    return f"Command({self.text!r})"

read(context)

Source code in src/anycfg/sources/_command.py

def read(self, context: "LoadContext") -> Payload:
    # A backstop: the loader refuses an executing source before it gets
    # here, but a source can also be read directly.
    from .._loader import check_command

    check_command(context.policy, self.text)
    timeout = (
        context.policy.command_timeout if self.timeout is _UNSET else self.timeout
    )
    codec = self.encoding or context.encoding
    if self._script is not None:
        raw = _run(_script_command(self._script), codec, timeout, shell=False)
    else:
        raw = _run(self.command, codec, timeout, shell=True)
    try:
        text = _decode(raw, codec, strict=True)
    except UnicodeDecodeError as error:
        failure = ParseError(
            f"output is not valid {codec} at byte {error.start}; pass "
            "encoding= naming the program's codec",
            path=self.text,
        )
    else:
        failure = None
    if failure is not None:
        # Outside the except block: the decode error holds the output.
        raise failure
    # Trailing whitespace and leading blank lines only: indentation on the
    # first line is document text (an indented YAML mapping).
    text = text.rstrip()
    while text.startswith("\n"):
        text = text[1:]

    from .. import formats

    named, text = formats.declared(text)
    found = self.format if self.format is not None else named
    if found is None and self._script is not None:
        stem = os.path.splitext(os.path.basename(self._script))[0]
        claimed = formats.find_for_filename(stem)
        found = claimed.name if claimed is not None else None
    # None when nothing named a format: the loader may still have one
    # (its own format=), and otherwise raises naming the ways to give one.
    return Payload(
        text,
        name=self.text,
        identity=f"command:{self.text}",
        format=found,
        encoding=codec,
    )

script(path, *, format=None, timeout=_UNSET, encoding=None) classmethod

A script file (.sh, .bat, .cmd, .ps1), run without a shell.

Source code in src/anycfg/sources/_command.py

@classmethod
def script(
    cls,
    path: "_ty.Union[str, os.PathLike]",
    *,
    format: "_ty.Optional[str]" = None,
    timeout: "_ty.Optional[float]" = _UNSET,
    encoding: "_ty.Optional[str]" = None,
) -> "Command":
    """A script file (``.sh``, ``.bat``, ``.cmd``, ``.ps1``), run without a shell."""
    text = os.fspath(path)
    source = cls(text, format=format, timeout=timeout, encoding=encoding)
    source.command = text
    source._script = text
    return source

Schemes

anycfg.sources.register_scheme(scheme, factory, *, replace=False)

Make <scheme>://... strings load through factory(text).

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

Source code in src/anycfg/sources/__init__.py

def register_scheme(
    scheme: str, factory: "_ty.Callable[[str], Source]", *, replace: bool = False
) -> None:
    """Make ``<scheme>://...`` strings load through ``factory(text)``.

    Raises `ValueError` when *scheme* is taken, unless *replace* is true.
    """
    global _schemes
    _ensure_loaded()
    key = scheme.lower()
    with _lock:
        if key in _schemes and not replace:
            raise ValueError(
                f"scheme {key!r} is already registered; pass replace=True to take it over"
            )
        _schemes = {**_schemes, key: (factory, _REGISTERED)}

anycfg.sources.unregister_scheme(scheme)

Remove scheme; return whether it was registered.

Source code in src/anycfg/sources/__init__.py

def unregister_scheme(scheme: str) -> bool:
    """Remove *scheme*; return whether it was registered."""
    global _schemes
    _ensure_loaded()
    with _lock:
        table = dict(_schemes)
        found = table.pop(scheme.lower(), None) is not None
        _schemes = table
        return found

anycfg.sources.schemes()

Every registered scheme, sorted.

Source code in src/anycfg/sources/__init__.py

def schemes() -> "_ty.List[str]":
    """Every registered scheme, sorted."""
    _ensure_loaded()
    return sorted(_schemes)

anycfg.sources.from_text(text, *, allow_plugins=True)

The source text names, or None when it is not a registered scheme.

allow_plugins false limits the lookup to built-in and code-registered schemes, which is what a document is allowed to reach.

Source code in src/anycfg/sources/__init__.py

def from_text(text: str, *, allow_plugins: bool = True) -> "_ty.Optional[Source]":
    """The source *text* names, or `None` when it is not a registered scheme.

    *allow_plugins* false limits the lookup to built-in and code-registered
    schemes, which is what a document is allowed to reach.
    """
    match = SCHEME_REGEX.match(text)
    if match is None:
        return None
    _ensure_loaded()
    found = _schemes.get(match.group(1).lower())
    if found is None:
        return None
    factory, origin = found
    if origin == _ENTRY_POINT and not allow_plugins:
        return None
    return factory(text)

anycfg.sources.SCHEME_REGEX = re.compile('^([A-Za-z][A-Za-z0-9.\\-]*?)(?:\\+([A-Za-z0-9_.\\-]+))?://') module-attribute

anycfg.sources.ENTRY_POINT_GROUP = 'anycfg.sources' module-attribute