"""
Configuration Utilities
Load and validate configuration settings from multiple sources with a
deterministic fallback order:
1. an explicit JSON or YAML file (or the first valid one in a folder),
2. one or more ``.env`` files merged into ``os.environ``,
3. plain environment variables.
The helpers verify that every required key is present and raise a clear
``RuntimeError`` when none of the sources can satisfy the request.
Author:
- Warith HARCHAOUI, https://linkedin.com/in/warith-harchaoui
"""
# Defer annotation evaluation so ``dict | None`` unions parse on Python 3.10.
from __future__ import annotations
import glob
import json
import os
# PyYAML is loaded lazily-but-eagerly here because YAML is one of the two
# accepted on-disk config formats (the other being JSON).
import yaml
from dotenv import dotenv_values, load_dotenv
from .logging_utils import error, info
from .path_utils import checkfile, dir_exists, file_exists, folder_name_ext, join
from .string_utils import emptystring
def _valid_config_file(a_path: str, keys: list[str], config_type: str) -> dict | None:
"""
Check if a configuration file is valid by verifying it contains the required keys.
This function loads a configuration file (in JSON or YAML format) and checks if it contains
all the specified keys.
You should not use it directly.
Parameters
----------
a_path : str
The path to the configuration file.
keys : List[str]
List of keys that must be present in the configuration file.
config_type : str
The type of configuration (for logging purposes).
Returns
-------
Optional[Dict]
The loaded configuration dictionary if valid, otherwise None.
Example
-------
>>> valid_config_file("config.yaml", ["host", "port"], "database")
{'host': 'localhost', 'port': 5432}
"""
# Fail early if the file is missing — the caller passed us a path it
# believed existed, so a missing file is a hard error, not a soft miss.
checkfile(a_path, f"Configuration file for {config_type} does not exist: {a_path}")
# Dispatch on the file extension to pick the right parser.
_, _, ext = folder_name_ext(a_path)
ext = ext.lower()
config = None
with open(a_path) as fin:
# JSON and YAML are the only two supported on-disk formats.
if "json" in ext:
config = json.load(fin)
elif any(e in ext for e in ["yaml", "yml"]):
# ``SafeLoader`` refuses arbitrary object construction — never load
# untrusted YAML with the full loader.
config = yaml.load(fin, Loader=yaml.SafeLoader)
# An unrecognized extension leaves ``config`` as None: treat it as a soft
# miss so the caller can fall through to the next config source.
if config is None:
info(f"Unsupported configuration file format {ext}: {a_path}")
return None
# A syntactically valid JSON/YAML document that isn't a mapping (a bare
# scalar or a list at the top level) is not a usable config: `key in
# config` below would either raise (a scalar isn't iterable) or silently
# check list membership instead of key presence. Treat it the same as an
# unsupported format -- a soft miss, not a crash.
if not isinstance(config, dict):
info(f"Configuration file '{a_path}' does not contain a mapping at the top level.")
return None
# A file only counts as valid when it satisfies EVERY required key.
if all(key in config for key in keys):
info(f"Configuration '{config_type}' successfully loaded from '{a_path}'")
return config
else:
# Otherwise report exactly which keys are missing to aid debugging.
missing = [key for key in keys if key not in config]
m = ", ".join(missing)
info(f"Configuration file '{a_path}' does not have all keys. Missing keys are {m}")
return None
def _config_from_mapping(
keys: list[str], mapping: dict[str, str]
) -> dict[str, str | int | float] | None:
"""
Extract configuration from an arbitrary str-to-str mapping.
Shared lookup logic behind both :func:`_config_from_env` (mapping =
``os.environ``) and the HTTP API's restricted, ambient-environment-free
resolution (mapping = the merged contents of the requested ``.env``
files only, via :func:`dotenv.dotenv_values`, never ``os.environ``
itself). You should not use it directly.
Parameters
----------
keys : List[str]
Keys to retrieve from ``mapping``.
mapping : Dict[str, str]
The key/value source to search.
Returns
-------
Optional[Dict[str, Union[str, int, float]]]
A dictionary of key-value pairs if all keys are found, otherwise None.
"""
config = {}
missing_keys = []
for key in keys:
cap_key = key.upper()
# Environment variables are conventionally UPPER_CASE, so try that
# spelling first...
if cap_key in mapping:
config[key] = mapping[cap_key]
# ...then fall back to the exact-case key for callers who set it verbatim.
elif key in mapping:
config[key] = mapping[key]
else:
# Track misses so we can report all of them at once below.
missing_keys.append(key)
# Only a fully-satisfied key set counts as a successful load.
if len(missing_keys) == 0:
return config
# Partial matches are treated as a miss so the caller can try another source.
m = ", ".join(missing_keys)
info(f"Missing keys: {m}")
return None
def _config_from_env(keys: list[str]) -> dict[str, str | int | float] | None:
"""
Extract configuration from system environment variables.
You should not use it directly.
Parameters
----------
keys : List[str]
Keys to retrieve from the environment.
Returns
-------
Optional[Dict[str, Union[str, int, float]]]
A dictionary of key-value pairs if all keys are found, otherwise None.
"""
return _config_from_mapping(keys, os.environ)
[docs]
def get_config(
keys: list[str],
config_type: str,
path: str | None = None,
env_files: list[str] | None = None,
*,
allow_ambient_env: bool = True,
) -> dict[str, str | int | float]:
"""
Load configuration settings using a fixed fallback order.
Precedence (first match wins):
1. ``path`` pointing to a JSON/YAML file (or, if a directory, the first
``.json``, ``.yaml`` or ``.yml`` file in it that contains all keys);
2. one or more ``.env`` files merged into ``os.environ``;
3. the current process environment (skipped when ``allow_ambient_env``
is ``False``).
Parameters
----------
keys : List[str]
Keys required to be present in the resolved configuration.
config_type : str
Human-readable label used only in log messages.
path : Optional[str], optional
Path to a configuration file or a directory containing one. If None
or empty, this step is skipped.
env_files : Optional[List[str]], optional
``.env`` files to load into ``os.environ`` before reading variables.
Defaults to ``[".env"]``.
allow_ambient_env : bool, optional
When ``True`` (the default, and the only behavior before this
parameter existed), step 3 reads the live process environment —
which includes both the requested ``.env`` files *and* whatever the
process happened to inherit at start-up. When ``False``, step 3 is
restricted to exactly the requested ``env_files``' own contents
(read directly, never merged into or read back from ``os.environ``):
a key that resolves only because it happens to be set in the
ambient environment is treated as unresolved. Used by the HTTP API
(see :mod:`os_helper.api`), where "the caller can name any
environment-variable key and get its live value back" is a
credential-exposure shape, not a feature — a local CLI/library
caller already has that access by definition, so this only ever
needs to be ``False`` across a network boundary.
Returns
-------
Dict[str, Union[str, int, float]]
Mapping with one entry per requested key.
Raises
------
RuntimeError
If none of the sources provide all required keys.
Example
-------
>>> config = get_config(["host", "port"], "database", path="config.yaml")
>>> config
{'host': 'localhost', 'port': 5432}
"""
if env_files is None:
env_files = [".env"]
# Step 1: Attempt to load from a specific file or folder if `path` is provided
info(f"Loading configuration for '{config_type}'")
config = None
if not emptystring(path):
if file_exists(path):
config = _valid_config_file(path, keys, config_type)
if config is not None:
info(f"Configuration '{config_type}' loaded from file: {path}")
return config
if dir_exists(path):
ext = ["json", "yaml", "yml"]
candidates = []
for e in ext:
candidates.extend(glob.glob(join(path, f"*.{e}")))
candidates = sorted(candidates)
for candidate_path in candidates:
config = _valid_config_file(candidate_path, keys, config_type)
if config is not None:
info(f"Configuration '{config_type}' loaded from file: {candidate_path}")
return config
info(f"No valid configuration found in path: {path}")
if allow_ambient_env:
# Step 2: Merge all .env files into os.environ
info("Loading configuration from env files")
for env_file in env_files:
if file_exists(env_file):
load_dotenv(env_file)
info(f"Loaded env file: {env_file}")
# Step 3: Check os.environ (files + whatever the process inherited) for required keys
info("Loading configuration from environment variables (possibly merged with .env files)")
config = _config_from_env(keys)
else:
# Step 2+3, restricted: read each env_file's own contents directly
# (dotenv_values never touches os.environ), so a key can only
# resolve here because a requested file actually defines it — never
# because the process happened to inherit it from its own shell.
info("Loading configuration from env files only (ambient process environment excluded)")
merged: dict[str, str] = {}
for env_file in env_files:
if file_exists(env_file):
merged.update(dotenv_values(env_file))
info(f"Loaded env file: {env_file}")
config = _config_from_mapping(keys, merged)
if config is not None:
info(f"Configuration '{config_type}' successfully loaded from environment variables.")
return config
# Step 4: Raise if no configuration was found
msg = (
f"Missing required keys for '{config_type}' configuration "
f"in files, .env files, or environment variables."
)
error(msg)
raise RuntimeError(msg)