feat(settings): improve config loading logic

Load settings from .toml files, .env and environment variables. Also ensures a ROOT_PATH environment variable is
set. If ROOT_PATH is not set and no root_path argument is passed, the current working directory is used as root.
Settings paths can be a single .toml file, a folder containing .toml files or a list of .toml files and folders.
If a folder is passed, all .toml files in the folder are loaded. If settings path is None, only .env and
environment variables are loaded. If settings_path are relative paths, they are joined with the root_path argument.

pyinfra/config/loader.py

@@ -1,5 +1,6 @@
 import argparse
 import os
+from functools import partial
 from pathlib import Path
 from typing import Union
 
@@ -7,20 +8,27 @@ from dynaconf import Dynaconf, ValidationError, Validator
 from funcy import lflatten
 from kn_utils.logging import logger
 
+# This path is ment for testing purposes and convenience. It probably won't reflect the actual root path when pyinfra is
+# installed as a package, so don't use it in production code, but define your own root path as described in load config.
+local_pyinfra_root_path = Path(__file__).parents[2]
 
-def load_settings(settings_path: Union[str, Path] = None, validators: list[Validator] = None):
-    settings_path = Path(settings_path) if settings_path else None
-    validators = validators or get_all_validators()
 
-    if not settings_path:
-        logger.info("No settings path specified, only loading .env end ENVs.")
-        settings_files = []
-    elif os.path.isdir(settings_path):
-        logger.info(f"Settings path is a directory, loading all .toml files in the directory: {settings_path}")
-        settings_files = list(settings_path.glob("*.toml"))
-    else:
-        logger.info(f"Settings path is a file, loading only the specified file: {settings_path}")
-        settings_files = [settings_path]
+def load_settings(
+    settings_path: Union[str, Path, list] = "config/",
+    root_path: Union[str, Path] = None,
+    validators: list[Validator] = None,
+):
+    """Load settings from .toml files, .env and environment variables. Also ensures a ROOT_PATH environment variable is
+    set. If ROOT_PATH is not set and no root_path argument is passed, the current working directory is used as root.
+    Settings paths can be a single .toml file, a folder containing .toml files or a list of .toml files and folders.
+    If a folder is passed, all .toml files in the folder are loaded. If settings path is None, only .env and
+    environment variables are loaded. If settings_path are relative paths, they are joined with the root_path argument.
+    """
+
+    root_path = get_or_set_root_path(root_path)
+    validators = validators or get_pyinfra_validators()
+
+    settings_files = normalize_to_settings_files(settings_path, root_path)
 
     settings = Dynaconf(
         load_dotenv=True,
@@ -34,10 +42,70 @@ def load_settings(settings_path: Union[str, Path] = None, validators: list[Valid
     return settings
 
 
-pyinfra_config_path = Path(__file__).resolve().parents[2] / "config/"
+def normalize_to_settings_files(settings_path: Union[str, Path, list], root_path: Union[str, Path]):
+    if settings_path is None:
+        logger.info("No settings path specified, only loading .env end ENVs.")
+        settings_files = []
+    elif isinstance(settings_path, str) or isinstance(settings_path, Path):
+        settings_files = [settings_path]
+    elif isinstance(settings_path, list):
+        settings_files = settings_path
+    else:
+        raise ValueError(f"Invalid settings path: {settings_path=}")
+
+    settings_files = lflatten(map(partial(_normalize_and_verify, root_path=root_path), settings_files))
+
+    return settings_files
 
 
-def get_all_validators():
+def _normalize_and_verify(settings_path: Path, root_path: Path):
+    settings_path = Path(settings_path)
+    root_path = Path(root_path)
+
+    if not settings_path.is_absolute():
+        logger.debug(f"Settings path is not absolute, joining with root path: {root_path}")
+        settings_path = root_path / settings_path
+
+    if settings_path.is_dir():
+        logger.debug(f"Settings path is a directory, loading all .toml files in the directory: {settings_path}")
+        settings_files = list(settings_path.glob("*.toml"))
+    elif settings_path.is_file():
+        logger.debug(f"Settings path is a file, loading specified file: {settings_path}")
+        settings_files = [settings_path]
+    else:
+        raise ValueError(f"Invalid settings path: {settings_path=}, {root_path=}")
+
+    return settings_files
+
+
+def get_or_set_root_path(root_path: Union[str, Path] = None):
+    env_root_path = os.environ.get("ROOT_PATH")
+
+    if env_root_path and root_path:
+        if Path(env_root_path) != Path(root_path):
+            logger.warning(
+                f"'ROOT_PATH' environment variable is set to {env_root_path}, but a different root_path argument was passed. "
+                f"Setting new value {root_path}."
+            )
+            os.environ["ROOT_PATH"] = str(root_path)
+
+    elif env_root_path:
+        root_path = env_root_path
+        logger.debug(f"'ROOT_PATH' environment variable is set to {root_path}.")
+
+    elif root_path:
+        logger.info(f"'ROOT_PATH' environment variable is not set, setting to {root_path}.")
+        os.environ["ROOT_PATH"] = str(root_path)
+
+    else:
+        root_path = Path.cwd()
+        logger.info(f"'ROOT_PATH' environment variable is not set, defaulting to working directory {root_path}.")
+        os.environ["ROOT_PATH"] = str(root_path)
+
+    return root_path
+
+
+def get_pyinfra_validators():
     import pyinfra.config.validators
 
     return lflatten(
@@ -64,10 +132,8 @@ def validate_settings(settings: Dynaconf, validators):
 def parse_args():
     parser = argparse.ArgumentParser()
     parser.add_argument(
-        "--settings_path",
-        "-s",
-        type=Path,
-        default=pyinfra_config_path,
-        help="Path to settings file or folder. Must be a .toml file or a folder containing .toml files.",
+        "settings_path",
+        help="Path to settings file(s) or folder(s). Must be .toml file(s) or a folder(s) containing .toml files.",
+        nargs="+",
     )
     return parser.parse_args()

pyinfra/examples.py

@@ -2,7 +2,7 @@ from dynaconf import Dynaconf
 from fastapi import FastAPI
 from kn_utils.logging import logger
 
-from pyinfra.config.loader import get_all_validators, validate_settings
+from pyinfra.config.loader import get_pyinfra_validators, validate_settings
 from pyinfra.queue.callback import Callback
 from pyinfra.queue.manager import QueueManager
 from pyinfra.utils.opentelemetry import instrument_pika, setup_trace
@@ -28,7 +28,7 @@ def start_standard_queue_consumer(
     Workload is received via queue messages and processed by the callback function (see pyinfra.queue.callback for
     callbacks).
     """
-    validate_settings(settings, get_all_validators())
+    validate_settings(settings, get_pyinfra_validators())
 
     logger.info(f"Starting webserver and queue consumer...")
 

scripts/send_request.py

@@ -4,11 +4,11 @@ from operator import itemgetter
 
 from kn_utils.logging import logger
 
-from pyinfra.config.loader import load_settings, pyinfra_config_path
+from pyinfra.config.loader import load_settings, local_pyinfra_root_path
 from pyinfra.queue.manager import QueueManager
 from pyinfra.storage.storages.s3 import get_s3_storage_from_settings
 
-settings = load_settings(pyinfra_config_path)
+settings = load_settings(local_pyinfra_root_path / "config/")
 
 
 def upload_json_and_make_message_body():

tests/conftest.py

@@ -2,14 +2,14 @@ import json
 
 import pytest
 
-from pyinfra.config.loader import load_settings, pyinfra_config_path
+from pyinfra.config.loader import load_settings, local_pyinfra_root_path
 from pyinfra.queue.manager import QueueManager
 from pyinfra.storage.connection import get_storage
 
 
 @pytest.fixture(scope="session")
 def settings():
-    return load_settings(pyinfra_config_path)
+    return load_settings(local_pyinfra_root_path / "config/")
 
 
 @pytest.fixture(scope="class")

tests/unit_test/config_test.py

@@ -1,9 +1,10 @@
 import os
+from pathlib import Path
 
 import pytest
 from dynaconf import Validator
 
-from pyinfra.config.loader import load_settings
+from pyinfra.config.loader import load_settings, local_pyinfra_root_path, normalize_to_settings_files
 from pyinfra.config.validators import webserver_validators
 
 
@@ -22,7 +23,7 @@ class TestConfig:
 
         validators = webserver_validators
 
-        test_settings = load_settings(validators=validators)
+        test_settings = load_settings(root_path=local_pyinfra_root_path, validators=validators)
 
         assert test_settings.webserver.host == "localhost"
 
@@ -30,7 +31,25 @@ class TestConfig:
         os.environ["TEST__VALUE__INT"] = "1"
         os.environ["TEST__VALUE__STR"] = "test"
 
-        test_settings = load_settings(validators=test_validators)
+        test_settings = load_settings(root_path=local_pyinfra_root_path, validators=test_validators)
 
         assert test_settings.test.value.int == 1
         assert test_settings.test.value.str == "test"
+
+    @pytest.mark.parametrize(
+        "settings_path,expected_file_paths",
+        [
+            (None, []),
+            ("config", [f"{local_pyinfra_root_path}/config/settings.toml"]),
+            ("config/settings.toml", [f"{local_pyinfra_root_path}/config/settings.toml"]),
+            (f"{local_pyinfra_root_path}/config", [f"{local_pyinfra_root_path}/config/settings.toml"]),
+        ],
+    )
+    def test_normalize_settings_files(self, settings_path, expected_file_paths):
+        files = normalize_to_settings_files(settings_path, local_pyinfra_root_path)
+        print(files)
+
+        assert len(files) == len(expected_file_paths)
+
+        for path, expected in zip(files, expected_file_paths):
+            assert path == Path(expected).absolute()