Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.rst
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,10 @@ Added
- New ``extension_or_*`` parser modes, e.g. ``extension_or_yaml``, which load
each config file in the format given by its extension, and otherwise as the
given mode (`#1021 <https://github.com/mauvilsa/jsonargparse/pull/1021>`__).
- Config file paths accept a ``#key.path`` suffix to parse only that part of the
file, e.g. ``pyproject.toml#tool.myapp``, in default config files, the config
argument and ``parse_path``. Default config files without the key are skipped
(`#1022 <https://github.com/mauvilsa/jsonargparse/pull/1022>`__).

Fixed
^^^^^
Expand Down
10 changes: 10 additions & 0 deletions DOCUMENTATION.rst
Original file line number Diff line number Diff line change
Expand Up @@ -1378,6 +1378,16 @@ All matching files are parsed in the given order and override the defaults from
the source code. They are always parsed first, so any command line argument
overrides their values.

Only part of a config file can be parsed by appending ``#`` and a dot separated
key, e.g. ``pyproject.toml#tool.myapp`` or ``package.json#myapp``, for files
shared with other tools. This works for default config files, the config
argument and :py:meth:`.ArgumentParser.parse_path`. A default config file that
does not have the key is skipped, otherwise it is an error. A selected part
cannot load another part of the same file. Combined with an ``extension_or_``
parser mode, an app can read ``pyproject.toml`` and also its own YAML config,
e.g.
``default_config_files=['pyproject.toml#tool.myapp', '.myapp.yaml']``.

An argument can also be added to give a config file path explicitly. This does
not disable ``default_config_files``. The config argument is parsed at its
position among the command line arguments, so arguments after it override the
Expand Down
10 changes: 7 additions & 3 deletions jsonargparse/_actions.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
)
from ._loaders_dumpers import get_loader_exceptions, load_value
from ._namespace import Namespace, copy_provenance
from ._optionals import _get_config_read_mode, ruamel_support
from ._optionals import ruamel_support
from ._paths import cwd_relative_path
from ._type_checking import ArgumentParser
from ._util import (
Expand All @@ -29,6 +29,7 @@
argument_error,
config_file_context,
default_config_option_help,
get_config_path,
get_import_path,
import_object,
indent_text,
Expand Down Expand Up @@ -127,8 +128,10 @@ def apply_config(parser, cfg, dest, value) -> None:

with parser_context(single_subcommand=False), previous_config_context(cfg), skip_apply_links():
kwargs = {"env": False, "defaults": False, "_skip_validation": True, "_fail_no_subcommand": False}
cfg_path: Path | None
selector: str | None = None
try:
cfg_path: Path | None = Path(value, mode=_get_config_read_mode())
cfg_path, selector = get_config_path(value)
except TypeError as ex_path:
try:
try:
Expand All @@ -151,7 +154,8 @@ def apply_config(parser, cfg, dest, value) -> None:
copy_provenance(cfg_merged, cfg)
if cfg.get(dest) is get_parsing_setting("unset_sentinel"):
cfg[dest] = []
cfg[dest].append(None if cfg_path is None else cwd_relative_path(cfg_path))
config_path = None if cfg_path is None else cwd_relative_path(cfg_path)
cfg[dest].append(f"{config_path}#{selector}" if selector else config_path)

def completer(self, prefix, **kwargs):
from ._completions import get_files_completer
Expand Down
73 changes: 49 additions & 24 deletions jsonargparse/_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -108,18 +108,21 @@
)
from ._util import (
ComposedConfig,
ConfigKeyNotFoundError,
Path,
argument_error,
check_no_composed_config,
config_file_context,
get_argument_group_class,
get_config_path,
get_private_kwargs,
identity,
indent_text,
load_config_path_context,
merge_config,
resolve_config_includes,
return_parser_if_captured,
select_config_key,
)

__all__ = ["ArgumentParser"]
Expand Down Expand Up @@ -338,7 +341,8 @@
``jsonnet``, ``omegaconf`` or one added via :func:`.set_loader`. With an ``extension_or_``
prefix, config files are loaded in the format given by their extension.
dump_header: Header to include as comment when dumping a config object.
default_config_files: Default config file locations, e.g. ``['~/.config/myapp/*.yaml']``.
default_config_files: Default config file locations, e.g. ``['~/.config/myapp/*.yaml']``, optionally
with a ``#key.path`` to parse only that part, e.g. ``pyproject.toml#tool.myapp``.
default_env: Whether to parse environment variables by default.
allow_abbrev: Whether to accept abbreviated long options, ``None`` to use the global setting.
"""
Expand Down Expand Up @@ -737,7 +741,8 @@
"""Parses a configuration file given its path.

Args:
path: Path to the configuration file to parse.
path: Path to the configuration file to parse, optionally with a ``#key.path`` to parse only that
part, e.g. ``pyproject.toml#tool.myapp``.
ext_vars: External variables for parsing Jsonnet.
env: Whether to merge with the parsed environment, ``None`` to use the parser's default.
defaults: Whether to merge with the parser's defaults.
Expand All @@ -749,16 +754,16 @@
ArgumentError: If the parsing fails and ``exit_on_error=False``.
TypeError: If the path does not exist or is not readable.
"""
fpath = Path(path, mode=_get_config_read_mode())
with load_config_path_context(fpath), path_dir_context(fpath):
fpath, selector = get_config_path(path)
Comment thread
mauvilsa marked this conversation as resolved.
Resolved
with load_config_path_context(fpath, selector), path_dir_context(fpath):
content = fpath.read_text()
parsed_cfg = self.parse_string(
content=content,
path=os.path.basename(urlsplit(fpath.absolute).path if fpath.is_url else path),
path=os.path.basename(urlsplit(fpath.absolute).path if fpath.is_url else fpath.relative),
ext_vars=ext_vars,
env=env,
defaults=defaults,
_source=ValueSource("config file", fpath, self.parser_mode),
_source=ValueSource("config file", fpath, self.parser_mode, selector=selector),
**kwargs,
)

Expand Down Expand Up @@ -795,7 +800,9 @@

try:
with parser_context(load_value_mode=self.parser_mode), value_source_context(source):
cfg = self._load_config_parser_mode(content, path, ext_vars, previous_config.get())
cfg = self._load_config_parser_mode(
content, path, ext_vars, previous_config.get(), selector=source.selector
)

if defaults or env:
cfg_base = self._parse_defaults_and_environ(defaults, env)
Expand All @@ -821,23 +828,28 @@
path: str | os.PathLike = "",
ext_vars: dict | None = None,
prev_cfg: Namespace | None = None,
selector: str | None = None,
) -> Namespace:
"""Loads a configuration string into a namespace.

Args:
content: The configuration content.
path: Optional path to original config path, just for error printing.
ext_vars: Optional external variables used for parsing jsonnet.
selector: Optional ``key.path`` of the part of the config to load.

Raises:
TypeError: If there is an invalid value according to the parser.
ConfigKeyNotFoundError: If the selected key is not in the config.
"""
try:
cfg_dict = load_config_value(content, path, path=path, ext_vars=ext_vars)
except get_loader_exceptions() as ex:
raise TypeError(f"Problems parsing config:\n{indent_text(str(ex))}") from ex
if not isinstance(cfg_dict, dict):
raise TypeError(f"Unexpected config: {content}")
if selector is not None:
cfg_dict = select_config_key(cfg_dict, selector)
return self._apply_actions(resolve_config_includes(cfg_dict), prev_cfg=prev_cfg)

## Methods for adding to the parser ##
Expand Down Expand Up @@ -1124,18 +1136,24 @@

## Methods related to defaults ##

def _get_default_config_files(self) -> list[Path]:
def _get_default_config_files(self) -> list[tuple[Path, str | None]]:
"""Returns the existing default config files, each with the key selected in it with ``path#key``."""
if getattr(self, "_inner_parser", False):
return []

default_config_files = []
default_config_files: list[tuple[str, str | None]] = []

for pattern in self.default_config_files:
default_config_files += sorted(glob.glob(os.path.expanduser(pattern)))
matches = sorted(glob.glob(os.path.expanduser(pattern)))
file_pattern, _, selector = pattern.rpartition("#")
if matches or not (file_pattern and selector):
default_config_files += [(m, None) for m in matches]
else:
default_config_files += [(m, selector) for m in sorted(glob.glob(os.path.expanduser(file_pattern)))]

if len(default_config_files) > 0:
with suppress(TypeError):
return [Path(v, mode=_get_config_read_mode()) for v in default_config_files]
return [(Path(v, mode=_get_config_read_mode()), s) for v, s in default_config_files]
return []

def get_default(self, dest: str) -> Any:
Expand Down Expand Up @@ -1164,7 +1182,7 @@
check_suppressed_default()
return defaults.get(action.dest)

def get_defaults(self, skip_validation: bool = False) -> Namespace:

Check failure on line 1185 in jsonargparse/_core.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this function to reduce its Cognitive Complexity from 17 to the 15 allowed.

See more on https://sonarcloud.io/project/issues?id=mauvilsa_jsonargparse&issues=AaFcrW7tlS_oY7n54HtZ&open=AaFcrW7tlS_oY7n54HtZ&pullRequest=1022
"""Returns a namespace with all default values.

Args:
Expand All @@ -1187,20 +1205,28 @@
self._logger.debug("Loaded parser defaults: %s", cfg)

default_config_files = self._get_default_config_files()
for default_config_file in default_config_files:
for default_config_file, selector in default_config_files:
name = cwd_relative_path(default_config_file) + (f"#{selector}" if selector else "")
with (
load_config_path_context(default_config_file),
load_config_path_context(default_config_file, selector),
path_dir_context(default_config_file),
parser_context(parent_parser=self, parsing_defaults=True),
):
default_config_file_content = default_config_file.read_text()
if not default_config_file_content.strip():
continue
source = ValueSource("default config file", default_config_file, self.parser_mode)
source = ValueSource("default config file", default_config_file, self.parser_mode, selector=selector)
with value_source_context(source):
cfg_file = self._load_config_parser_mode(
default_config_file_content, os.path.basename(default_config_file), prev_cfg=cfg
)
try:
cfg_file = self._load_config_parser_mode(
default_config_file_content,
os.path.basename(default_config_file),
prev_cfg=cfg,
selector=selector,
)
except ConfigKeyNotFoundError as ex:
self._logger.debug("Skipped default config file %s: %s", name, ex)
continue
cfg = merge_config(self, cfg_file, cfg)
try:
with _ActionPrintConfig.skip_print_config():
Expand All @@ -1214,18 +1240,17 @@
)
except (TypeError, KeyError, argparse.ArgumentError) as ex:
raise argument_error(
f"Problem in default config file '{default_config_file}': {ex.args[0]}",
default_config_file=str(default_config_file),
f"Problem in default config file '{name}': {ex.args[0]}",
default_config_file=str(name),
) from ex
meta = cfg.get("__default_config__")
config_path = cwd_relative_path(default_config_file)
if isinstance(meta, list):
meta.append(config_path)
meta.append(name)
elif isinstance(meta, str):
cfg["__default_config__"] = [meta, config_path]
cfg["__default_config__"] = [meta, name]
else:
cfg["__default_config__"] = config_path
self._logger.debug("Parsed default configuration from path: %s", default_config_file)
cfg["__default_config__"] = name
self._logger.debug("Parsed default configuration from path: %s", name)

with parser_context(validating_defaults=True):
ActionTypeHint.add_sub_defaults(self, cfg)
Expand Down
4 changes: 3 additions & 1 deletion jsonargparse/_formatters.py
Original file line number Diff line number Diff line change
Expand Up @@ -346,6 +346,8 @@ def describe_source(source, cache: dict | None = None) -> str:
if source.implied_by is not None:
return f"{source.description}, from {describe_source(source.implied_by, cache)}"
text = source.description if source.origin is None else f"{source.description} {_describe_origin(source.origin)}"
if source.selector:
text += f"#{source.selector}"
position = _find_position(source, {} if cache is None else cache)
return text if position is None else f"{text}:{position[0]}"

Expand Down Expand Up @@ -380,7 +382,7 @@ def _find_position(source, cache: dict) -> tuple[int, int, str] | None:
content = source.origin.read_text()
cache[id(source.origin)] = (_get_config_key_positions(content), content.splitlines())
positions, lines = cache[id(source.origin)]
key = source.key
key = f"{source.selector}.{source.key}" if source.selector else source.key
while key:
for candidate in [key, ("." + key).replace(".init_args.", ".")[1:]]: # init_args can be implicit
if candidate in positions:
Expand Down
1 change: 1 addition & 0 deletions jsonargparse/_namespace.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ class ValueSource(NamedTuple):
mode: str | None = None # the parser mode of a config file, required to find line numbers
key: str | None = None # the key in a config file that the value corresponds to
implied_by: "ValueSource | None" = None # for implicit values, e.g. a class_path, the source of what implied it
selector: str | None = None # the key selected in a config file with path#key, of which key is relative

def __deepcopy__(self, memo):
return self # immutable, so copies of namespaces can share it, which is much faster, e.g. for list items
Expand Down
62 changes: 52 additions & 10 deletions jsonargparse/_util.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@


default_config_option_help = "Path to a configuration file."
config_load_stack: ContextVar[tuple[tuple[str, str], ...]] = ContextVar("config_load_stack", default=())
config_load_stack: ContextVar[tuple[tuple[str, str, str | None], ...]] = ContextVar("config_load_stack", default=())


def argument_error(message: str, default_config_file: str | None = None) -> ArgumentError:
Expand Down Expand Up @@ -76,32 +76,74 @@ def merge_config(parser, source: Namespace, target: Namespace) -> Namespace:
return target


def _config_path_id(cfg_path: Path) -> tuple[str, str]:
class ConfigKeyNotFoundError(TypeError):
"""Raised when the key selected in a config with ``path#key`` is not in it."""


def get_config_path(value: str | os.PathLike) -> tuple[Path, str | None]:
"""Returns the path of a config file and the key selected in it with ``path#key.path``, if any.

The value is split on the last ``#`` only if it is not a path itself but the part before is.

Raises:
TypeError: If the value is not a path to a config file.
"""
mode = _get_config_read_mode()
try:
return Path(value, mode=mode), None
except TypeError as ex:
path, _, selector = os.fspath(value).rpartition("#")
if not path or not selector:
raise
try:
return Path(path, mode=mode), selector
except TypeError:
raise ex from None


def select_config_key(cfg: dict, selector: str) -> dict:
"""Returns the mapping at a ``key.path`` selected in a loaded config."""
value: Any = cfg
for key in selector.split("."):
if not isinstance(value, dict) or key not in value:
raise ConfigKeyNotFoundError(f'Key "{selector}" not found in the config')
value = value[key]
if not isinstance(value, dict):
raise TypeError(f'Expected a mapping at key "{selector}". Got value: {value}')
return value


def _config_path_id(cfg_path: Path, selector: str | None) -> tuple[str, str, str | None]:
path_id = cfg_path.absolute
if not (cfg_path.is_url or cfg_path.is_fsspec):
path_id = os.path.realpath(path_id)
return path_id, str(cfg_path)
return path_id, str(cfg_path) + (f"#{selector}" if selector else ""), selector


def _format_config_load_chain(stack: tuple[tuple[str, str], ...], path_id: tuple[str, str]) -> str:
def _format_config_load_chain(
stack: tuple[tuple[str, str, str | None], ...], path_id: tuple[str, str, str | None]
) -> str:
chain = list(stack) + [path_id]
for num, (stack_path, _) in enumerate(chain):
for num, (stack_path, _, _) in enumerate(chain):
if stack_path == path_id[0]:
chain = chain[num:]
break
return " -> ".join(display for _, display in chain)
return " -> ".join(display for _, display, _ in chain)


@contextmanager
def load_config_path_context(cfg_path: Path | None) -> Iterator[None]:
def load_config_path_context(cfg_path: Path | None, selector: str | None = None) -> Iterator[None]:
if cfg_path is None:
yield
return
path_id = _config_path_id(cfg_path)
path_id = _config_path_id(cfg_path, selector)
stack = config_load_stack.get()
if path_id[0] in {path for path, _ in stack}:
selectors = {sel for path, _, sel in stack if path == path_id[0]}
if selectors:
chain = _format_config_load_chain(stack, path_id)
raise TypeError(f"Config file loop detected: {chain}")
if selector in selectors:
raise TypeError(f"Config file loop detected: {chain}")
raise TypeError(f"Loading another section of the same config file is not supported: {chain}")
token = config_load_stack.set(stack + (path_id,))
try:
yield
Expand Down
Loading
Loading