405 lines
17 KiB
Python
405 lines
17 KiB
Python
import io
|
||
import json
|
||
import warnings
|
||
from configparser import (
|
||
MAX_INTERPOLATION_DEPTH,
|
||
ConfigParser,
|
||
ExtendedInterpolation,
|
||
InterpolationDepthError,
|
||
InterpolationMissingOptionError,
|
||
InterpolationSyntaxError,
|
||
NoOptionError,
|
||
NoSectionError,
|
||
ParsingError,
|
||
)
|
||
from typing import Any, Dict, List
|
||
|
||
from ._constants import SECTION_PREFIX
|
||
from ._errors import ConfigValidationError
|
||
from .util import VARIABLE_RE, try_dump_json, try_load_json
|
||
|
||
|
||
def parse_config(
|
||
text: str,
|
||
*,
|
||
interpolate: bool = True,
|
||
overrides: Dict[str, Any] = {},
|
||
) -> dict[str, Any]:
|
||
"""Parse a config string into a nested dict.
|
||
|
||
Handles the full pipeline: parse with ConfigParser, validate structure,
|
||
apply overrides, interpret values, and resolve section references.
|
||
|
||
Returns the nested dict and whether a second interpolation pass is needed
|
||
(when overrides were applied with interpolation enabled).
|
||
"""
|
||
config_parser = _get_configparser(interpolate=interpolate and not overrides)
|
||
try:
|
||
config_parser.read_string(text)
|
||
except ParsingError as e:
|
||
desc = f"Make sure the sections and values are formatted correctly.\n\n{e}"
|
||
raise ConfigValidationError(desc=desc) from None
|
||
errors = _validate_configparser(config_parser)
|
||
if errors:
|
||
raise errors[0]
|
||
errors = _validate_overrides(config_parser, overrides)
|
||
if errors:
|
||
raise errors[0]
|
||
result: dict[str, Any] = {}
|
||
section_parts = [section.split(".") for section in config_parser.sections()]
|
||
# Phase 1: Build the skeleton of nested dicts from section names.
|
||
for parts in section_parts:
|
||
node = result
|
||
for part in parts[:-1]:
|
||
node = node.setdefault(part, {}) if part == "*" else node[part]
|
||
node.setdefault(parts[-1], {})
|
||
# Phase 2: Fill in values, processing breadth-first by section depth.
|
||
for section, values in sorted(
|
||
config_parser.items(), key=lambda x: len(x[0].split("."))
|
||
):
|
||
if section == "DEFAULT":
|
||
continue
|
||
parts = section.split(".")
|
||
node = result
|
||
for part in parts:
|
||
node = node[part]
|
||
for key in values:
|
||
node[key] = _interpret_value(config_parser.get(section, key))
|
||
# Phase 3: Apply overrides on the nested dict.
|
||
_apply_overrides(result, overrides)
|
||
# Phase 4: Replace section reference placeholders with actual dicts.
|
||
_replace_section_refs(result, result)
|
||
return result
|
||
|
||
|
||
def _apply_overrides(result: dict[str, Any], overrides: Dict[str, Any]) -> None:
|
||
"""Apply dot-notation overrides to a nested dict.
|
||
|
||
Override paths have already been validated by _validate_overrides.
|
||
"""
|
||
for key, value in overrides.items():
|
||
path = key.split(".")
|
||
node = result
|
||
for part in path[:-1]:
|
||
node = node[part]
|
||
node[path[-1]] = value
|
||
|
||
|
||
def serialize_config(
|
||
data: dict[str, Any],
|
||
*,
|
||
interpolate: bool = True,
|
||
inline_paths: frozenset[str] = frozenset(),
|
||
) -> str:
|
||
"""Serialize a nested config dict to a config string.
|
||
|
||
inline_paths: dotted paths whose values should be serialized as inline
|
||
JSON rather than expanded into subsections. For example, if "a.b" is
|
||
in inline_paths, data["a"]["b"] (even if it's a dict) will be
|
||
serialized as ``b = {"key": "value"}`` under [a].
|
||
"""
|
||
flattened = _get_configparser(interpolate=interpolate)
|
||
queue: list[tuple[tuple, dict[str, Any]]] = [(tuple(), data)]
|
||
for path, node in queue:
|
||
section_name = ".".join(path)
|
||
is_kwarg = path and path[-1] != "*"
|
||
has_leaves = any(not hasattr(v, "items") for v in node.values())
|
||
if path and has_leaves and not flattened.has_section(section_name):
|
||
# Create sections that have leaf values (including * sections).
|
||
flattened.add_section(section_name)
|
||
elif is_kwarg and not flattened.has_section(section_name):
|
||
flattened.add_section(section_name)
|
||
for key, value in node.items():
|
||
child_path = f"{section_name}.{key}" if section_name else key
|
||
if hasattr(value, "items") and child_path not in inline_paths:
|
||
queue.append((path + (key,), value))
|
||
else:
|
||
flattened.set(section_name, key, try_dump_json(value, node))
|
||
string_io = io.StringIO()
|
||
flattened.write(string_io)
|
||
return string_io.getvalue().strip()
|
||
|
||
|
||
def _get_configparser(interpolate: bool = True) -> ConfigParser:
|
||
config = ConfigParser(interpolation=_CustomInterpolation() if interpolate else None)
|
||
# Preserve case of keys: https://stackoverflow.com/a/1611877/6400719
|
||
config.optionxform = str # type: ignore
|
||
return config
|
||
|
||
|
||
def _validate_configparser(config_parser: ConfigParser) -> list[ConfigValidationError]:
|
||
"""Validate a configparser's structure before interpreting it into a Config.
|
||
|
||
Checks that:
|
||
- No values leak into the DEFAULT section (top-level values without a section)
|
||
- All parent sections exist for dotted section names (e.g. "a.b" requires "a")
|
||
- No key in a section conflicts with a child section name
|
||
- No interpolation errors in values
|
||
"""
|
||
errors = []
|
||
default_section = config_parser.defaults()
|
||
if default_section: # pragma: no cover -- configparser raises ParsingError first
|
||
err_title = "Found config values without a top-level section"
|
||
err_msg = "not part of a section"
|
||
err = [{"loc": [k], "msg": err_msg} for k in default_section]
|
||
errors.append(ConfigValidationError(errors=err, title=err_title))
|
||
section_names = set(config_parser.sections())
|
||
for section in config_parser.sections():
|
||
path = section.split(".")
|
||
for i in range(1, len(path)):
|
||
# "*" is an implicit list section — it doesn't need a parent
|
||
# section header, and paths through it are always valid.
|
||
if path[i - 1] == "*":
|
||
continue
|
||
parent = ".".join(path[:i])
|
||
# A parent is valid if it's a declared section OR if the path
|
||
# goes through a "*" component (which is implicitly created).
|
||
if parent not in section_names and "*" not in parent.split("."):
|
||
err_title = (
|
||
"Error parsing config section. Perhaps a section name is wrong?"
|
||
)
|
||
err = [{"loc": path, "msg": f"Section '{path[i - 1]}' is not defined"}]
|
||
errors.append(ConfigValidationError(errors=err, title=err_title))
|
||
break
|
||
keys = set(config_parser.options(section))
|
||
for other in section_names:
|
||
if other.startswith(section + "."):
|
||
child = other[len(section) + 1 :].split(".")[0]
|
||
if child in keys:
|
||
err = [{"loc": other.split("."), "msg": "found conflicting values"}]
|
||
errors.append(ConfigValidationError(errors=err))
|
||
return errors
|
||
|
||
|
||
def _validate_overrides(
|
||
config_parser: ConfigParser, overrides: dict[str, Any]
|
||
) -> list[ConfigValidationError]:
|
||
errors = []
|
||
err_title = "Error parsing config overrides"
|
||
for key in overrides:
|
||
err_msg = "not a section value that can be overridden"
|
||
err = [{"loc": key.split("."), "msg": err_msg}]
|
||
if "." not in key:
|
||
errors.append(ConfigValidationError(errors=err, title=err_title))
|
||
continue
|
||
section, _ = key.rsplit(".", 1)
|
||
# Check for section and accept if option not in config[section]
|
||
if section not in config_parser:
|
||
errors.append(ConfigValidationError(errors=err, title=err_title))
|
||
# TODO: Are we supposed to chek for the *option*?
|
||
return errors
|
||
|
||
|
||
def _interpret_value(value: Any) -> Any:
|
||
"""Interpret a single config value."""
|
||
result = try_load_json(value)
|
||
# If value is a string and it contains a variable, use original value
|
||
# (not interpreted string, which could lead to double quotes:
|
||
# ${x.y} -> "${x.y}" -> "'${x.y}'"). Make sure to check it's a string,
|
||
# so we're not keeping lists as strings.
|
||
# NOTE: This currently can't handle uninterpolated values like [${x.y}]!
|
||
if isinstance(result, str) and VARIABLE_RE.search(value):
|
||
result = value
|
||
return result
|
||
|
||
|
||
def _replace_section_refs(
|
||
root: dict[str, Any], node: dict[str, Any], parent: str = ""
|
||
) -> None:
|
||
"""Replace section reference placeholders with actual dicts."""
|
||
for key, value in node.items():
|
||
key_parent = f"{parent}.{key}".strip(".")
|
||
if isinstance(value, dict):
|
||
_replace_section_refs(root, value, parent=key_parent)
|
||
elif isinstance(value, list):
|
||
node[key] = [_get_section_ref(root, v, parent=[parent, key]) for v in value]
|
||
else:
|
||
node[key] = _get_section_ref(root, value, parent=[parent, key])
|
||
|
||
|
||
def _get_section_ref(
|
||
root: Dict[str, Any], value: Any, *, parent: List[str] = []
|
||
) -> Any:
|
||
"""Resolve a single section reference placeholder, or return value as-is."""
|
||
if isinstance(value, str) and value.startswith(
|
||
f'"{SECTION_PREFIX}'
|
||
): # pragma: no cover
|
||
value = try_load_json(value) # pragma: no cover
|
||
if (
|
||
isinstance(value, str)
|
||
and value.startswith(SECTION_PREFIX)
|
||
and value != SECTION_PREFIX
|
||
):
|
||
parts = value.replace(SECTION_PREFIX, "", 1).split(".")
|
||
result = root
|
||
for item in parts:
|
||
result = result[item]
|
||
return result
|
||
elif isinstance(value, str) and SECTION_PREFIX in value and value != SECTION_PREFIX:
|
||
err_desc = (
|
||
"Can't reference whole sections or return values of function "
|
||
"blocks inside a string or list\n\nYou can change your variable to "
|
||
"reference a value instead. Keep in mind that it's not "
|
||
"possible to interpolate the return value of a registered "
|
||
"function, since variables are interpolated when the config "
|
||
"is loaded, and registered functions are resolved afterwards."
|
||
)
|
||
err = [{"loc": parent, "msg": "uses section variable in string or list"}]
|
||
raise ConfigValidationError(errors=err, desc=err_desc)
|
||
return value
|
||
|
||
|
||
class _CustomInterpolation(ExtendedInterpolation):
|
||
def before_read( # pyright: ignore[reportIncompatibleMethodOverride]
|
||
self, parser: ConfigParser, section: str, option: str, value: str
|
||
) -> str:
|
||
# Warn about single-quoted strings (common mistake)
|
||
if value and value[0] == value[-1] == "'":
|
||
warnings.warn(
|
||
f"The value [{value}] seems to be single-quoted, but values "
|
||
"use JSON formatting, which requires double quotes."
|
||
)
|
||
return super().before_read(parser, section, option, value)
|
||
|
||
def _coerce_for_string_context(self, v: str) -> str:
|
||
"""Coerce a raw config value for use in a compound string expression."""
|
||
# Don't coerce section references - they need to stay quoted for JSON
|
||
if SECTION_PREFIX in v:
|
||
return v
|
||
try:
|
||
parsed = json.loads(v)
|
||
except json.JSONDecodeError:
|
||
return v # Not valid JSON, already a plain string
|
||
if isinstance(parsed, str):
|
||
return parsed # Unwrap JSON string
|
||
# Use json.dumps() for non-strings, escaping inner quotes so they don't
|
||
# conflict with the outer JSON string quotes
|
||
return json.dumps(parsed).replace('"', '\\"')
|
||
|
||
def before_get( # pyright: ignore[reportIncompatibleMethodOverride]
|
||
self,
|
||
parser: ConfigParser,
|
||
section: str,
|
||
option: str,
|
||
value: str,
|
||
defaults: Dict[str, str],
|
||
) -> str:
|
||
# Mostly copy-pasted from the built-in configparser implementation.
|
||
# The interpolate() method resolves ${...} references and appends pieces
|
||
# to L. For a bare reference like ${x}, L has one element. For compound
|
||
# expressions like "hello ${x}", L has multiple pieces that we join.
|
||
# Compound results stay as strings (coerced via _coerce_for_string_context),
|
||
# while bare references keep their JSON type for _interpret_value to parse.
|
||
L: List[str] = []
|
||
self.interpolate(parser, option, L, value, section, defaults, 1)
|
||
if len(L) == 1:
|
||
return L[0]
|
||
return "".join(self._coerce_for_string_context(piece) for piece in L)
|
||
|
||
def interpolate(
|
||
self,
|
||
parser: ConfigParser,
|
||
option: str,
|
||
accum: List[str],
|
||
rest: str,
|
||
section: str,
|
||
map: Dict[str, str],
|
||
depth: int,
|
||
) -> None:
|
||
"""Resolve variable references like ${foo.bar}"""
|
||
# Mostly copy-pasted from the built-in configparser implementation.
|
||
# We need to overwrite this method so we can add special handling for
|
||
# block references :( All values produced here should be strings –
|
||
# we need to wait until the whole config is interpreted anyways so
|
||
# filling in incomplete values here is pointless. All we need is the
|
||
# section reference so we can fetch it later.
|
||
rawval = parser.get(section, option, raw=True, fallback=rest)
|
||
if depth > MAX_INTERPOLATION_DEPTH:
|
||
raise InterpolationDepthError(option, section, rawval)
|
||
while rest:
|
||
p = rest.find("$")
|
||
if p < 0:
|
||
accum.append(rest)
|
||
return
|
||
if p > 0:
|
||
accum.append(rest[:p])
|
||
rest = rest[p:]
|
||
# p is no longer used
|
||
c = rest[1:2]
|
||
if c == "$":
|
||
accum.append("$")
|
||
rest = rest[2:]
|
||
elif c == "{":
|
||
# We want to treat both ${a:b} and ${a.b} the same
|
||
m = self._KEYCRE.match(rest) # type: ignore[attr-defined]
|
||
if m is None:
|
||
err = f"bad interpolation variable reference {rest}"
|
||
raise InterpolationSyntaxError(option, section, err)
|
||
orig_var = m.group(1)
|
||
path = orig_var.replace(":", ".").rsplit(".", 1)
|
||
rest = rest[m.end() :]
|
||
sect = section
|
||
opt = option
|
||
try:
|
||
if len(path) == 1:
|
||
opt = parser.optionxform(path[0])
|
||
# Check if the variable references a section rather
|
||
# than a key in the current section. If the key
|
||
# exists in the current map but its raw value is the
|
||
# same interpolation variable (self-reference), or if
|
||
# the key doesn't exist in the map, treat it as a
|
||
# section reference.
|
||
is_section_ref = opt not in map
|
||
if not is_section_ref:
|
||
raw = map[opt]
|
||
is_section_ref = raw.strip() == rawval.strip()
|
||
if not is_section_ref:
|
||
v = map[opt]
|
||
else:
|
||
# We have block reference, store it as a special key
|
||
section_name = parser[parser.optionxform(path[0])].name
|
||
v = self._get_section_name(section_name)
|
||
elif len(path) == 2:
|
||
sect = path[0]
|
||
opt = parser.optionxform(path[1])
|
||
fallback = "__FALLBACK__"
|
||
v = parser.get(sect, opt, raw=True, fallback=fallback)
|
||
# If a variable doesn't exist, try again and treat the
|
||
# reference as a section
|
||
if v == fallback:
|
||
v = self._get_section_name(parser[f"{sect}.{opt}"].name)
|
||
else: # pragma: no cover
|
||
# Dead code: rsplit(".", 1) produces at most 2 elements
|
||
err = f"More than one ':' found: {rest}"
|
||
raise InterpolationSyntaxError(option, section, err)
|
||
except (KeyError, NoSectionError, NoOptionError):
|
||
raise InterpolationMissingOptionError(
|
||
option, section, rawval, orig_var
|
||
) from None
|
||
if "$" in v:
|
||
new_map = dict(parser.items(sect, raw=True))
|
||
self.interpolate(parser, opt, accum, v, sect, new_map, depth + 1)
|
||
else:
|
||
accum.append(v)
|
||
else:
|
||
err = "'$' must be followed by '$' or '{', found: %r" % (rest,)
|
||
raise InterpolationSyntaxError(option, section, err)
|
||
|
||
def _get_section_name(self, name: str) -> str:
|
||
"""Generate the name of a section. Note that we use a quoted string here
|
||
so we can use section references within lists and load the list as
|
||
JSON. Since section references can't be used within strings, we don't
|
||
need the quoted vs. unquoted distinction like we do for variables.
|
||
|
||
Examples (assuming section = {"foo": 1}):
|
||
- value: ${section.foo} -> value: 1
|
||
- value: "hello ${section.foo}" -> value: "hello 1"
|
||
- value: ${section} -> value: {"foo": 1}
|
||
- value: "${section}" -> value: {"foo": 1}
|
||
- value: "hello ${section}" -> invalid
|
||
"""
|
||
return f'"{SECTION_PREFIX}{name}"'
|
||
|
||
|
||
__all__ = ["parse_config", "serialize_config"]
|