Configuration sources (Effect's ConfigProvider).
A ConfigProvider looks a path up and finds a Node: a Value
(a string), a Record (child keys) or an Array (a length), or
None when nothing is there::
provider = config_provider.from_env(env={"APP_PORT": "3000"})
provider.load(("APP", "PORT")) # Effect[Node | None, SourceError]: Value("3000")Sources: from_env (os.environ by default), from_env_record,
from_unknown (nested dicts and lists), from_dot_env_contents /
from_dot_env (.env files), from_dir (a file per key) and
make (any lookup function). Combinators are methods: or_else,
map_input, nested, constant_case. CurrentConfigProvider is
the provider configuration is read from (from_env() unless a layer
or layer_add installs another).
Classes#
Functions#
from_dir#
from_dir(root_path='/', *, preserve_empty_strings=False) -> ConfigProviderA provider over the files under root_path: the path ("db", "password") reads the file <root_path>/db/password (trimmed) as a
value, and a directory is a record of its entry names. A missing path is
absent; any other read error (a permission error, say) fails with a
SourceError. An empty file is absent unless preserve_empty_strings.
Files are read on a worker thread at each lookup.
| Parameter | Type | Description |
|---|---|---|
root_path | str | os.PathLike[str] = '/' | |
preserve_empty_strings | bool = False |
Returns ConfigProvider
def from_dir(
root_path: str | os.PathLike[str] = "/", *, preserve_empty_strings: bool = False
) -> ConfigProvider:
"""A provider over the files under ``root_path``: the path ``("db",
"password")`` reads the file ``<root_path>/db/password`` (trimmed) as a
value, and a directory is a record of its entry names. A missing path is
absent; any other read error (a permission error, say) fails with a
``SourceError``. An empty file is absent unless ``preserve_empty_strings``.
Files are read on a worker thread at each lookup."""
root = os.fspath(root_path)
return make(lambda path: _load(root, path, preserve_empty_strings))from_dot_env#
from_dot_env(path='.env', *, expand_variables=False, preserve_empty_strings=False) -> Effect[ConfigProvider, SourceError]Read path (UTF-8, on a worker thread) and build
from_dot_env_contents from it. A file that cannot be read fails with
a SourceError whose cause is the OSError.
| Parameter | Type | Description |
|---|---|---|
path | str | Path = '.env' | |
expand_variables | bool = False | |
preserve_empty_strings | bool = False |
Returns Effect[ConfigProvider, SourceError]
def from_dot_env(
path: str | Path = ".env",
*,
expand_variables: bool = False,
preserve_empty_strings: bool = False,
) -> Effect[ConfigProvider, SourceError]:
"""Read ``path`` (UTF-8, on a worker thread) and build
``from_dot_env_contents`` from it. A file that cannot be read fails with
a ``SourceError`` whose ``cause`` is the ``OSError``."""
return try_blocking(
lambda: Path(path).read_text(encoding="utf-8"),
lambda error: SourceError(f"Failed to read file at {path}", error),
).map(
lambda contents: from_dot_env_contents(
contents,
expand_variables=expand_variables,
preserve_empty_strings=preserve_empty_strings,
)
)from_dot_env_contents#
from_dot_env_contents(contents, *, expand_variables=False, preserve_empty_strings=False) -> ConfigProviderA provider over the NAME=value lines of a .env file, read as
from_env_record reads them. # comments, export prefixes and
single, double or backtick quotes are understood; \n and \r
are expanded inside double quotes. With expand_variables, $NAME,
${NAME} and ${NAME:-default} are replaced by other entries of the
file (\$ stays a literal $).
| Parameter | Type | Description |
|---|---|---|
contents | str | |
expand_variables | bool = False | |
preserve_empty_strings | bool = False |
Returns ConfigProvider
def from_dot_env_contents(
contents: str, *, expand_variables: bool = False, preserve_empty_strings: bool = False
) -> ConfigProvider:
"""A provider over the ``NAME=value`` lines of a ``.env`` file, read as
``from_env_record`` reads them. ``#`` comments, ``export`` prefixes and
single, double or backtick quotes are understood; ``\\n`` and ``\\r``
are expanded inside double quotes. With ``expand_variables``, ``$NAME``,
``${NAME}`` and ``${NAME:-default}`` are replaced by other entries of the
file (``\\$`` stays a literal ``$``)."""
env = _parse(contents)
if expand_variables:
env = _expand(env)
return from_env_record(env, preserve_empty_strings=preserve_empty_strings)from_env#
from_env(env=None, *, preserve_empty_strings=False) -> ConfigProviderfrom_env_record over env, or over os.environ (names and
values read at each lookup) when no env is given.
| Parameter | Type | Description |
|---|---|---|
env | Mapping[str, str] | None = None | |
preserve_empty_strings | bool = False |
Returns ConfigProvider
def from_env(
env: Mapping[str, str] | None = None, *, preserve_empty_strings: bool = False
) -> ConfigProvider:
"""``from_env_record`` over ``env``, or over ``os.environ`` (names and
values read at each lookup) when no ``env`` is given."""
if env is not None:
return from_env_record(env, preserve_empty_strings=preserve_empty_strings)
return make(
lambda path: succeed(_node_at(_trie(os.environ), os.environ, path, preserve_empty_strings))
)from_env_record#
from_env_record(env, *, preserve_empty_strings=False) -> ConfigProviderA provider over env; None values are skipped. A name is split
on every _ (A__B has an empty middle segment), and a path also
matches the name its segments join to, so ("NODE_ENV",) and
("NODE", "ENV") both find NODE_ENV. A path whose children are
all array indices (A_0, A_1) is an Array; one with other
children a Record, holding the path's own value if it has one. An
empty value is absent unless preserve_empty_strings. As in Effect,
the names are read when the provider is made and the values at each
lookup.
| Parameter | Type | Description |
|---|---|---|
env | Mapping[str, str | None] | |
preserve_empty_strings | bool = False |
Returns ConfigProvider
def from_env_record(
env: Mapping[str, str | None], *, preserve_empty_strings: bool = False
) -> ConfigProvider:
"""A provider over ``env``; ``None`` values are skipped. A name is split
on every ``_`` (``A__B`` has an empty middle segment), and a path also
matches the name its segments join to, so ``("NODE_ENV",)`` and
``("NODE", "ENV")`` both find ``NODE_ENV``. A path whose children are
all array indices (``A_0``, ``A_1``) is an ``Array``; one with other
children a ``Record``, holding the path's own value if it has one. An
empty value is absent unless ``preserve_empty_strings``. As in Effect,
the names are read when the provider is made and the values at each
lookup."""
trie = _trie(env)
return make(lambda path: succeed(_node_at(trie, env, path, preserve_empty_strings)))layer#
layer(provider) -> Layer[CurrentConfigProvider, object, object]Install provider, or the provider an effect builds, replacing the
current one.
| Parameter | Type | Description |
|---|---|---|
provider | ConfigProvider | Effect[ConfigProvider, object, object] |
Returns Layer[CurrentConfigProvider, object, object]
def layer(
provider: ConfigProvider | Effect[ConfigProvider, object, object], /
) -> Layer[CurrentConfigProvider, object, object]:
"""Install ``provider``, or the provider an effect builds, replacing the
current one."""
if isinstance(provider, ConfigProvider):
return _layer.succeed(CurrentConfigProvider, provider)
return _layer.effect(CurrentConfigProvider, _provider_effect(provider))layer_add#
layer_add(provider, /, *, as_primary=False) -> Layer[CurrentConfigProvider, object, object]Combine provider with the current one: as its fallback, or, with
as_primary, consulted first.
| Parameter | Type | Description |
|---|---|---|
provider | ConfigProvider | Effect[ConfigProvider, object, object] | |
as_primary | bool = False |
Returns Layer[CurrentConfigProvider, object, object]
def layer_add(
provider: ConfigProvider | Effect[ConfigProvider, object, object],
/,
*,
as_primary: bool = False,
) -> Layer[CurrentConfigProvider, object, object]:
"""Combine ``provider`` with the current one: as its fallback, or, with
``as_primary``, consulted first."""
added = _provider_effect(provider)
def combine(current: ConfigProvider, new: ConfigProvider) -> ConfigProvider:
return new.or_else(current) if as_primary else current.or_else(new)
combined = CurrentConfigProvider().flat_map(
lambda current: added.map(lambda new: combine(current, new))
)
return _layer.effect(CurrentConfigProvider, combined)make#
make(get) -> ConfigProviderA provider over get, which receives the transformed path and
returns the node there, None, or a SourceError failure.
| Parameter | Type | Description |
|---|---|---|
get | Lookup |
Returns ConfigProvider
def make(get: Lookup) -> ConfigProvider:
"""A provider over ``get``, which receives the transformed path and
returns the node there, ``None``, or a ``SourceError`` failure."""
return _Source(get, _identity)from_unknown#
from_unknown(root, *, preserve_empty_strings=False) -> ConfigProviderA provider over nested mappings, lists and tuples. Mappings are
records (string segments), lists and tuples arrays (int segments); a
string is a value, as are numbers and booleans ("42", "true");
None is absent. An empty string is absent unless
preserve_empty_strings.
| Parameter | Type | Description |
|---|---|---|
root | object | |
preserve_empty_strings | bool = False |
Returns ConfigProvider
def from_unknown(root: object, *, preserve_empty_strings: bool = False) -> ConfigProvider:
"""A provider over nested mappings, lists and tuples. Mappings are
records (string segments), lists and tuples arrays (int segments); a
string is a value, as are numbers and booleans (``"42"``, ``"true"``);
``None`` is absent. An empty string is absent unless
``preserve_empty_strings``."""
return make(lambda path: succeed(_node_at(root, path, preserve_empty_strings)))