18126767925

Committed 30 Sep 2025 10:25AM UTC coverage: 91.42% (-1.5%) from 92.881%

Build Type

push

github

Committed by Nick Bubelich

Commit Message

Add caching for file reading, improve substitution logic, and enhance error handling.

- Introduced `_read_file_cached` with `lru_cache

Run Details

41 of 51 new or added lines in 4 files covered. (80.39%)

4 existing lines in 1 file now uncovered.

586 of 641 relevant lines covered (91.42%)

0.91 hits per line

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

97.75

/tomlev/parser.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 functools import lru_cache
from pathlib import Path
from tomllib import loads as toml_loads
from typing import Any, TypeAlias

from .constants import DEFAULT_SEPARATOR
from .env_loader import EnvDict
from .errors import EnvironmentVariableError
from .include_handler import expand_includes_dict
from .patterns import RE_PATTERN

__all__ = ["ConfigDict", "read_toml", "substitute_and_parse"]

# Type aliases for clarity
ConfigDict: TypeAlias = dict[str, Any]


@lru_cache(maxsize=32)
def _read_file_cached(file_path: str) -> str:
    """Read file content with caching to avoid repeated disk I/O.

    Args:
        file_path: Path to the file to read.

    Returns:
        File content as string.
    """
    with io.open(file_path, mode="rt", encoding="utf8") as fp:
        return fp.read()


def substitute_and_parse(content: str, env: EnvDict, strict: bool, separator: str = DEFAULT_SEPARATOR) -> ConfigDict:
    """Substitute environment variables in content and parse TOML.

    Handles escapes (e.g., "$$" and "$1") and default syntax using the
    configured separator.

    Args:
        content: TOML content string with environment variable placeholders.
        env: Dictionary of environment variables for substitution.
        strict: Whether to operate in strict mode for error handling.
        separator: Separator string for default values in environment variables.

    Returns:
        Dictionary of parsed TOML configuration with substituted values.

    Raises:
        EnvironmentVariableError: In strict mode, when referenced variables are undefined.
    """
    # not found variables
    not_found_variables = set()

    # substitutions dictionary
    substitutions: dict[str, str] = {}

    # Build list of content segments for efficient string building
    segments: list[tuple[int, int, str]] = []  # (start, end, replacement)

    # iterate over findings
    for entry in RE_PATTERN.finditer(content):
        groups = entry.groupdict()

        # replace
        variable: str | None = None
        default: str | None = None
        replace: str | None = None

        match groups:
            case {"named": name, "named_default": def_val} if name:
                variable = name
                default = def_val
            case {"braced": name, "braced_default": def_val} if name:
                variable = name
                default = def_val
            case {"escaped": esc_val} if esc_val:
                span = entry.span()
                pref = groups.get("pref") or ""
                post = groups.get("post") or ""
                replacement = f"{pref}{esc_val}{post}"
                segments.append((span[0], span[1], replacement))
                continue

        if variable is not None:
            if variable in env:
                replace = env[variable]
            elif variable not in env and default is not None:
                replace = default
            else:
                not_found_variables.add(variable)

        if replace is not None and variable is not None:
            search = "${" if groups["braced"] else "$"
            search += variable
            if default is not None:
                search += separator + default
            search += "}" if groups["braced"] else ""
            substitutions[search] = replace

    if strict and not_found_variables:
        raise EnvironmentVariableError.missing_variables(list(not_found_variables))

    # Apply escape replacements efficiently using segments
    if segments:
        result_parts: list[str] = []
        last_end: int = 0
        for start, end, replacement in segments:
            result_parts.append(content[last_end:start])
            # Replacement is always str for escaped values, never None in practice
            if replacement is not None:
                result_parts.append(replacement)
            last_end = end
        result_parts.append(content[last_end:])
        content = "".join(result_parts)

    # Apply variable substitutions (single-pass, longest first to avoid partial replacements)
    for search in sorted(substitutions, key=len, reverse=True):
        content = content.replace(search, substitutions[search])

    # Parse TOML
    toml = toml_loads(content)
    if toml and isinstance(toml, dict):
        return toml
    return {}


def read_toml(file_path: str, env: EnvDict, strict: bool, separator: str = DEFAULT_SEPARATOR) -> ConfigDict:
    """Read and parse TOML file with environment variable substitution.

    Args:
        file_path: Path to the TOML file to read.
        env: Dictionary of environment variables for substitution.
        strict: Whether to operate in strict mode for error handling.
        separator: Separator string for default values in environment variables.

    Returns:
        Dictionary of parsed TOML configuration with substituted values.

    Raises:
        FileNotFoundError: When the specified TOML file doesn't exist.
        EnvironmentVariableError: In strict mode, when referenced variables are undefined.
    """
    # read file (with caching for repeated reads)
    try:
        content: str = _read_file_cached(file_path)
    except FileNotFoundError as e:
        raise FileNotFoundError(f"TOML file not found: {file_path}") from e
    except (OSError, IOError) as e:
        raise OSError(f"Error reading TOML file '{file_path}': {e}") from e

    # Perform substitution and parse
    try:
        toml = substitute_and_parse(content, env, strict, separator)
    except EnvironmentVariableError as e:
        # Add file context to environment variable errors
        errors = [(attr, f"{msg} (in file: {file_path})") for attr, msg in e.errors]
        raise EnvironmentVariableError(errors) from e
    except Exception as e:
        # Add file context to parsing errors
        raise ValueError(f"Error parsing TOML file '{file_path}': {e}") from e

    # Expand __include directives recursively
    if toml and isinstance(toml, dict):
        expand_includes_dict(
            toml,
            Path(file_path).parent,
            env,
            strict,
            separator,
            seen={Path(file_path).resolve()},
            cache={},
            substitute_and_parse_func=substitute_and_parse,
        )
        return toml

    return {}

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 functools import lru_cache	1✔
29	from pathlib import Path	1✔
30	from tomllib import loads as toml_loads	1✔
31	from typing import Any, TypeAlias	1✔
32
33	from .constants import DEFAULT_SEPARATOR	1✔
34	from .env_loader import EnvDict	1✔
35	from .errors import EnvironmentVariableError	1✔
36	from .include_handler import expand_includes_dict	1✔
37	from .patterns import RE_PATTERN	1✔
38
39	__all__ = ["ConfigDict", "read_toml", "substitute_and_parse"]	1✔
40
41	# Type aliases for clarity
42	ConfigDict: TypeAlias = dict[str, Any]	1✔
43
44
45	@lru_cache(maxsize=32)	1✔
46	def _read_file_cached(file_path: str) -> str:	1✔
47	"""Read file content with caching to avoid repeated disk I/O.
48
49	Args:
50	file_path: Path to the file to read.
51
52	Returns:
53	File content as string.
54	"""
55	with io.open(file_path, mode="rt", encoding="utf8") as fp:	1✔
56	return fp.read()	1✔
57
58
59	def substitute_and_parse(content: str, env: EnvDict, strict: bool, separator: str = DEFAULT_SEPARATOR) -> ConfigDict:	1✔
60	"""Substitute environment variables in content and parse TOML.
61
62	Handles escapes (e.g., "$$" and "$1") and default syntax using the
63	configured separator.
64
65	Args:
66	content: TOML content string with environment variable placeholders.
67	env: Dictionary of environment variables for substitution.
68	strict: Whether to operate in strict mode for error handling.
69	separator: Separator string for default values in environment variables.
70
71	Returns:
72	Dictionary of parsed TOML configuration with substituted values.
73
74	Raises:
75	EnvironmentVariableError: In strict mode, when referenced variables are undefined.
76	"""
77	# not found variables
78	not_found_variables = set()	1✔
79
80	# substitutions dictionary
81	substitutions: dict[str, str] = {}	1✔
82
83	# Build list of content segments for efficient string building
84	segments: list[tuple[int, int, str]] = [] # (start, end, replacement)	1✔
85
86	# iterate over findings
87	for entry in RE_PATTERN.finditer(content):	1✔
88	groups = entry.groupdict()	1✔
89
90	# replace
91	variable: str \| None = None	1✔
92	default: str \| None = None	1✔
93	replace: str \| None = None	1✔
94
95	match groups:	1✔
96	case {"named": name, "named_default": def_val} if name:	1✔
97	variable = name	1✔
98	default = def_val	1✔
99	case {"braced": name, "braced_default": def_val} if name:	1✔
100	variable = name	1✔
101	default = def_val	1✔
102	case {"escaped": esc_val} if esc_val:	1✔
103	span = entry.span()	1✔
104	pref = groups.get("pref") or ""	1✔
105	post = groups.get("post") or ""	1✔
106	replacement = f"{pref}{esc_val}{post}"	1✔
107	segments.append((span[0], span[1], replacement))	1✔
108	continue	1✔
109
110	if variable is not None:	1✔
111	if variable in env:	1✔
112	replace = env[variable]	1✔
113	elif variable not in env and default is not None:	1✔
114	replace = default	1✔
115	else:
116	not_found_variables.add(variable)	1✔
117
118	if replace is not None and variable is not None:	1✔
119	search = "${" if groups["braced"] else "$"	1✔
120	search += variable	1✔
121	if default is not None:	1✔
122	search += separator + default	1✔
123	search += "}" if groups["braced"] else ""	1✔
124	substitutions[search] = replace	1✔
125
126	if strict and not_found_variables:	1✔
127	raise EnvironmentVariableError.missing_variables(list(not_found_variables))	1✔
128
129	# Apply escape replacements efficiently using segments
130	if segments:	1✔
131	result_parts: list[str] = []	1✔
132	last_end: int = 0	1✔
133	for start, end, replacement in segments:	1✔
134	result_parts.append(content[last_end:start])	1✔
135	# Replacement is always str for escaped values, never None in practice
136	if replacement is not None:	1✔
137	result_parts.append(replacement)	1✔
138	last_end = end	1✔
139	result_parts.append(content[last_end:])	1✔
140	content = "".join(result_parts)	1✔
141
142	# Apply variable substitutions (single-pass, longest first to avoid partial replacements)
143	for search in sorted(substitutions, key=len, reverse=True):	1✔
144	content = content.replace(search, substitutions[search])	1✔
145
146	# Parse TOML
147	toml = toml_loads(content)	1✔
148	if toml and isinstance(toml, dict):	1✔
149	return toml	1✔
150	return {}	1✔
151
152
153	def read_toml(file_path: str, env: EnvDict, strict: bool, separator: str = DEFAULT_SEPARATOR) -> ConfigDict:	1✔
154	"""Read and parse TOML file with environment variable substitution.
155
156	Args:
157	file_path: Path to the TOML file to read.
158	env: Dictionary of environment variables for substitution.
159	strict: Whether to operate in strict mode for error handling.
160	separator: Separator string for default values in environment variables.
161
162	Returns:
163	Dictionary of parsed TOML configuration with substituted values.
164
165	Raises:
166	FileNotFoundError: When the specified TOML file doesn't exist.
167	EnvironmentVariableError: In strict mode, when referenced variables are undefined.
168	"""
169	# read file (with caching for repeated reads)
170	try:	1✔
171	content: str = _read_file_cached(file_path)	1✔
172	except FileNotFoundError as e:	1✔
173	raise FileNotFoundError(f"TOML file not found: {file_path}") from e	1✔
NEW 174	except (OSError, IOError) as e:	×
NEW 175	raise OSError(f"Error reading TOML file '{file_path}': {e}") from e	×
176
177	# Perform substitution and parse
178	try:	1✔
179	toml = substitute_and_parse(content, env, strict, separator)	1✔
180	except EnvironmentVariableError as e:	1✔
181	# Add file context to environment variable errors
182	errors = [(attr, f"{msg} (in file: {file_path})") for attr, msg in e.errors]	1✔
183	raise EnvironmentVariableError(errors) from e	1✔
184	except Exception as e:	1✔
185	# Add file context to parsing errors
186	raise ValueError(f"Error parsing TOML file '{file_path}': {e}") from e	1✔
187
188	# Expand __include directives recursively
189	if toml and isinstance(toml, dict):	1✔
190	expand_includes_dict(	1✔
191	toml,
192	Path(file_path).parent,
193	env,
194	strict,
195	separator,
196	seen={Path(file_path).resolve()},
197	cache={},
198	substitute_and_parse_func=substitute_and_parse,
199	)
200	return toml	1✔
201
202	return {}	1✔

thesimj / tomlev / 18126767925

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