update for dict support

Author: martvanrijthoven

README.md

@@ -67,9 +67,25 @@ pip install pyamlo
 ```python
 from pyamlo import load_config
 
+# Load from YAML file
 config = load_config("examples/test_config.yaml")
 print(config)
 
+# Load from Python dictionary
+config_dict = {
+    "app": {"name": "MyApp", "workers": 4},
+    "database": {"pool_size": "${app.workers * 2}"}
+}
+config = load_config(config_dict)
+print(config)
+
+# Load from multiple sources (files and dictionaries)
+config = load_config([
+    "base_config.yaml",
+    {"app": {"debug": True}},
+    "override_config.yaml"
+])
+print(config)
 ```
 See for more details on the [examples documentation](https://martvanrijthoven.github.io/pyamlo/examples/).
 

pyamlo/config.py

@@ -11,7 +11,7 @@
 
 
 def load_config(
-    source: Union[str, Path, IO[str], Sequence[Union[str, Path, IO[str]]]],
+    source: Union[str, Path, IO[str], dict, Sequence[Union[str, Path, IO[str], dict]]],
     overrides: Optional[list[str]] = None,
     use_cli: bool = False,
     security_policy: SecurityPolicy = SecurityPolicy(restrictive=False),

pyamlo/include.py

@@ -34,7 +34,9 @@ def load_raw(path: str) -> dict[str, Any]:
 
 
 def process_includes(
-    raw: dict[str, Any], base_path: str | None = None, security_policy: SecurityPolicy = None
+    raw: dict[str, Any],
+    base_path: str | None = None,
+    security_policy: SecurityPolicy = None,
 ) -> dict[str, Any]:
     incs = raw.pop("include!", [])
     merged: dict[str, Any] = {}

pyamlo/sources.py

@@ -1,12 +1,42 @@
 from pathlib import Path
 from typing import IO, Any, Sequence, Union
+import os
 
 from pyamlo.merge import deep_merge
 from pyamlo.include import process_includes, set_base_paths
-from pyamlo.tags import ConfigLoader
+from pyamlo.tags import ConfigLoader, CallSpec, ExtendSpec, PatchSpec
 from pyamlo.security import SecurityPolicy
 
 
+def _process_dict_tags(data: Any, security_policy: SecurityPolicy) -> Any:
+    """Process YAML-style tags in dictionary data."""
+    if isinstance(data, dict):
+        return {k: _process_dict_tags(v, security_policy) for k, v in data.items()}
+    elif isinstance(data, list):
+        return [_process_dict_tags(item, security_policy) for item in data]
+    elif isinstance(data, str):
+        # Check for YAML tags in string values
+        if data.startswith("!@"):
+            # Object instantiation tag
+            parts = data[2:].strip().split(None, 1)
+            path = parts[0]
+            args = [parts[1]] if len(parts) > 1 else []
+            return CallSpec(path, args, {}, is_interpolated=False)
+        elif data.startswith("!env "):
+            # Environment variable tag
+            var = data[5:].strip()
+            security_policy.check_env_var(var)
+            val = os.environ.get(var)
+            if val is None:
+                raise ValueError(f"Environment variable '{var}' not set")
+            return val
+        elif data == "!extend":
+            return ExtendSpec([])
+        elif data == "!patch":
+            return PatchSpec({})
+    return data
+
+
 def _load_source(
     source: Union[str, Path, IO[str]], security_policy: SecurityPolicy
 ) -> dict[str, Any]:
@@ -23,27 +53,31 @@ def _load_with_loader(stream):
 
 
 def _process_single_source(
-    src: Union[str, Path, IO[str]], security_policy: SecurityPolicy
+    src: Union[str, Path, IO[str], dict], security_policy: SecurityPolicy
 ) -> dict[str, Any]:
-    raw = _load_source(src, security_policy=security_policy)
+    if isinstance(src, dict):
+        raw = _process_dict_tags(src.copy(), security_policy)
+        src_path = "<dict>"
+    else:
+        raw = _load_source(src, security_policy=security_policy)
+        src_path = src.name if isinstance(src, IO) else str(src)
     if not raw:
         return {}
-    src_path = src.name if isinstance(src, IO) else str(src)
-    if src_path:
+    if src_path and src_path != "<dict>":
         set_base_paths(raw, src_path)
     return process_includes(raw, src_path, security_policy=security_policy)
 
 
 def get_sources(
-    source: Union[str, Path, IO[str], Sequence[Union[str, Path, IO[str]]]],
-) -> list[Union[str, Path, IO[str]]]:
+    source: Union[str, Path, IO[str], dict, Sequence[Union[str, Path, IO[str], dict]]],
+) -> list[Union[str, Path, IO[str], dict]]:
     if not isinstance(source, Sequence) or isinstance(source, (str, Path)):
         return [source]
     return list(source)
 
 
 def merge_all_sources(
-    sources: list[Union[str, Path, IO[str]]], security_policy: SecurityPolicy
+    sources: list[Union[str, Path, IO[str], dict]], security_policy: SecurityPolicy
 ) -> dict[str, Any]:
     config: dict[str, Any] = {}
     for src in sources:

pyamlo/tags.py

@@ -130,7 +130,7 @@ def construct_callspec(
     node: Union[MappingNode, SequenceNode, ScalarNode],
 ) -> CallSpec:
     args, kwargs = _construct_callspec_args(loader, node)
-    if '$' in suffix:
+    if "$" in suffix:
         return CallSpec(suffix, args, kwargs, is_interpolated=True)
     return CallSpec(suffix, args, kwargs, is_interpolated=False)