17697843040

Committed 13 Sep 2025 02:25PM UTC coverage: 91.945% (-3.1%) from 95.067%

Build Type

push

github

Committed by Nick Bubelich

Commit Message

Implement `__include` directive for nested TOML configurations and CLI validation tool.

Coverage Stats

424 of 465 new or added lines in 11 files covered. (91.18%)

468 of 509 relevant lines covered (91.94%)

0.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.67

/tomlev/include_handler.py

"""
MIT License

Copyright (c) 2025 Nick Bubelich

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
"""

from __future__ import annotations

import io
from pathlib import Path
from typing import Any

from .constants import INCLUDE_KEY
from .env_loader import EnvDict
from .errors import IncludeError

__all__ = ["deep_merge", "expand_includes_dict"]


def deep_merge(dst: dict[str, Any], src: dict[str, Any]) -> dict[str, Any]:
    """Deep-merge src into dst (dicts merge, scalars overwrite).

    Args:
        dst: Destination dictionary to merge into.
        src: Source dictionary to merge from.

    Returns:
        The merged destination dictionary.
    """
    for k, v in src.items():
        if k == INCLUDE_KEY:
            # never propagate include directive from included content
            continue
        if isinstance(v, dict) and isinstance(dst.get(k), dict):
            deep_merge(dst[k], v)
        else:
            dst[k] = v
    return dst


def expand_includes_dict(
    node: dict[str, Any],
    base_dir: Path,
    env: EnvDict,
    strict: bool,
    separator: str,
    *,
    seen: set[Path],
    cache: dict[Path, dict[str, Any]],
    substitute_and_parse_func: Any,  # Function to substitute and parse TOML content
) -> None:
    """Recursively expand __include directives within a parsed TOML dict.

    Args:
        node: The dictionary node to process for includes.
        base_dir: Base directory for resolving relative include paths.
        env: Environment variables dictionary for substitution.
        strict: Whether to operate in strict mode for error handling.
        separator: Separator string for default values in environment variables.
        seen: Set of already seen paths to detect cycles.
        cache: Cache of already processed include files.
        substitute_and_parse_func: Function to substitute variables and parse TOML.

    Raises:
        IncludeError: When include validation fails or cycles are detected.
        FileNotFoundError: In strict mode, when included files are not found.
    """
    # Normalize and process includes at current node
    if INCLUDE_KEY in node:
        raw = node.get(INCLUDE_KEY)
        includes: list[str]
        if isinstance(raw, str):
            includes = [raw]
        elif isinstance(raw, list) and all(isinstance(x, str) for x in raw):
            includes = list(raw)
        else:
            if strict:
                raise IncludeError.invalid_type()
            includes = []

        for rel in includes:
            include_path = (base_dir / rel).resolve()
            if include_path in seen:
                if strict:
                    raise IncludeError.cycle_detected(str(include_path))
                continue
            if not include_path.is_file():
                if strict:
                    raise FileNotFoundError(f"Included TOML not found: {include_path}")
                continue

            if include_path in cache:
                sub_dict = cache[include_path]
            else:
                with io.open(include_path, mode="rt", encoding="utf8") as fp:
                    sub_content = fp.read()
                sub_dict = substitute_and_parse_func(sub_content, env, strict, separator)
                # Recurse into the included dict for its own includes, carry seen
                expand_includes_dict(
                    sub_dict,
                    include_path.parent,
                    env,
                    strict,
                    separator,
                    seen=seen | {include_path},
                    cache=cache,
                    substitute_and_parse_func=substitute_and_parse_func,
                )
                cache[include_path] = sub_dict

            deep_merge(node, sub_dict)

        # Remove directive after processing
        node.pop(INCLUDE_KEY, None)

    # Recurse into children
    for k, v in list(node.items()):
        if isinstance(v, dict):
            expand_includes_dict(v, base_dir, env, strict, separator, seen=seen, cache=cache, substitute_and_parse_func=substitute_and_parse_func)

1	"""
2	MIT License
3
4	Copyright (c) 2025 Nick Bubelich
5
6	Permission is hereby granted, free of charge, to any person obtaining a copy
7	of this software and associated documentation files (the "Software"), to deal
8	in the Software without restriction, including without limitation the rights
9	to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
10	copies of the Software, and to permit persons to whom the Software is
11	furnished to do so, subject to the following conditions:
12
13	The above copyright notice and this permission notice shall be included in all
14	copies or substantial portions of the Software.
15
16	THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17	IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18	FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
19	AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20	LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
21	OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
22	SOFTWARE.
23	"""
24
25	from __future__ import annotations	1✔
26
27	import io	1✔
28	from pathlib import Path	1✔
29	from typing import Any	1✔
30
31	from .constants import INCLUDE_KEY	1✔
32	from .env_loader import EnvDict	1✔
33	from .errors import IncludeError	1✔
34
35	__all__ = ["deep_merge", "expand_includes_dict"]	1✔
36
37
38	def deep_merge(dst: dict[str, Any], src: dict[str, Any]) -> dict[str, Any]:	1✔
39	"""Deep-merge src into dst (dicts merge, scalars overwrite).
40
41	Args:
42	dst: Destination dictionary to merge into.
43	src: Source dictionary to merge from.
44
45	Returns:
46	The merged destination dictionary.
47	"""
48	for k, v in src.items():	1✔
49	if k == INCLUDE_KEY:	1✔
50	# never propagate include directive from included content
NEW 51	continue	×
52	if isinstance(v, dict) and isinstance(dst.get(k), dict):	1✔
53	deep_merge(dst[k], v)	1✔
54	else:
55	dst[k] = v	1✔
56	return dst	1✔
57
58
59	def expand_includes_dict(	1✔
60	node: dict[str, Any],
61	base_dir: Path,
62	env: EnvDict,
63	strict: bool,
64	separator: str,
65	*,
66	seen: set[Path],
67	cache: dict[Path, dict[str, Any]],
68	substitute_and_parse_func: Any, # Function to substitute and parse TOML content
69	) -> None:
70	"""Recursively expand __include directives within a parsed TOML dict.
71
72	Args:
73	node: The dictionary node to process for includes.
74	base_dir: Base directory for resolving relative include paths.
75	env: Environment variables dictionary for substitution.
76	strict: Whether to operate in strict mode for error handling.
77	separator: Separator string for default values in environment variables.
78	seen: Set of already seen paths to detect cycles.
79	cache: Cache of already processed include files.
80	substitute_and_parse_func: Function to substitute variables and parse TOML.
81
82	Raises:
83	IncludeError: When include validation fails or cycles are detected.
84	FileNotFoundError: In strict mode, when included files are not found.
85	"""
86	# Normalize and process includes at current node
87	if INCLUDE_KEY in node:	1✔
88	raw = node.get(INCLUDE_KEY)	1✔
89	includes: list[str]
90	if isinstance(raw, str):	1✔
91	includes = [raw]	1✔
92	elif isinstance(raw, list) and all(isinstance(x, str) for x in raw):	1✔
93	includes = list(raw)	1✔
94	else:
95	if strict:	1✔
96	raise IncludeError.invalid_type()	1✔
NEW 97	includes = []	×
98
99	for rel in includes:	1✔
100	include_path = (base_dir / rel).resolve()	1✔
101	if include_path in seen:	1✔
102	if strict:	1✔
103	raise IncludeError.cycle_detected(str(include_path))	1✔
NEW 104	continue	×
105	if not include_path.is_file():	1✔
106	if strict:	1✔
NEW 107	raise FileNotFoundError(f"Included TOML not found: {include_path}")	×
108	continue	1✔
109
110	if include_path in cache:	1✔
111	sub_dict = cache[include_path]	1✔
112	else:
113	with io.open(include_path, mode="rt", encoding="utf8") as fp:	1✔
114	sub_content = fp.read()	1✔
115	sub_dict = substitute_and_parse_func(sub_content, env, strict, separator)	1✔
116	# Recurse into the included dict for its own includes, carry seen
117	expand_includes_dict(	1✔
118	sub_dict,
119	include_path.parent,
120	env,
121	strict,
122	separator,
123	seen=seen \| {include_path},
124	cache=cache,
125	substitute_and_parse_func=substitute_and_parse_func,
126	)
127	cache[include_path] = sub_dict	1✔
128
129	deep_merge(node, sub_dict)	1✔
130
131	# Remove directive after processing
132	node.pop(INCLUDE_KEY, None)	1✔
133
134	# Recurse into children
135	for k, v in list(node.items()):	1✔
136	if isinstance(v, dict):	1✔
137	expand_includes_dict(v, base_dir, env, strict, separator, seen=seen, cache=cache, substitute_and_parse_func=substitute_and_parse_func)	1✔

thesimj / tomlev / 17697843040

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous