1"""A simple configuration system."""
2
3# Copyright (c) IPython Development Team.
4# Distributed under the terms of the Modified BSD License.
5from __future__ import annotations
6
7import argparse
8import copy
9import functools
10import json
11import os
12import re
13import sys
14import typing as t
15from logging import Logger
16
17from traitlets.traitlets import Any, Container, Dict, HasTraits, List, TraitType, Undefined
18
19from ..utils import cast_unicode, filefind, warnings
20
21# -----------------------------------------------------------------------------
22# Exceptions
23# -----------------------------------------------------------------------------
24
25
26class ConfigError(Exception):
27 pass
28
29
30class ConfigLoaderError(ConfigError):
31 pass
32
33
34class ConfigFileNotFound(ConfigError):
35 pass
36
37
38class ArgumentError(ConfigLoaderError):
39 pass
40
41
42# -----------------------------------------------------------------------------
43# Argparse fix
44# -----------------------------------------------------------------------------
45
46# Unfortunately argparse by default prints help messages to stderr instead of
47# stdout. This makes it annoying to capture long help screens at the command
48# line, since one must know how to pipe stderr, which many users don't know how
49# to do. So we override the print_help method with one that defaults to
50# stdout and use our class instead.
51
52
53class _Sentinel:
54 def __repr__(self) -> str:
55 return "<Sentinel deprecated>"
56
57 def __str__(self) -> str:
58 return "<deprecated>"
59
60
61_deprecated = _Sentinel()
62
63
64class ArgumentParser(argparse.ArgumentParser):
65 """Simple argparse subclass that prints help to stdout by default."""
66
67 def print_help(self, file: t.Any = None) -> None:
68 if file is None:
69 file = sys.stdout
70 return super().print_help(file)
71
72 print_help.__doc__ = argparse.ArgumentParser.print_help.__doc__
73
74
75# -----------------------------------------------------------------------------
76# Config class for holding config information
77# -----------------------------------------------------------------------------
78
79
80class LazyConfigValue(HasTraits):
81 """Proxy object for exposing methods on configurable containers
82
83 These methods allow appending/extending/updating
84 to add to non-empty defaults instead of clobbering them.
85
86 Exposes:
87
88 - append, extend, insert on lists
89 - update on dicts
90 - update, add on sets
91 """
92
93 _value = None
94
95 # list methods
96 _extend: List[t.Any] = List()
97 _prepend: List[t.Any] = List()
98 _inserts: List[t.Any] = List()
99
100 def append(self, obj: t.Any) -> None:
101 """Append an item to a List"""
102 self._extend.append(obj)
103
104 def extend(self, other: t.Any) -> None:
105 """Extend a list"""
106 self._extend.extend(other)
107
108 def prepend(self, other: t.Any) -> None:
109 """like list.extend, but for the front"""
110 self._prepend[:0] = other
111
112 def merge_into(self, other: t.Any) -> t.Any:
113 """
114 Merge with another earlier LazyConfigValue or an earlier container.
115 This is useful when having global system-wide configuration files.
116
117 Self is expected to have higher precedence.
118
119 Parameters
120 ----------
121 other : LazyConfigValue or container
122
123 Returns
124 -------
125 LazyConfigValue
126 if ``other`` is also lazy, a reified container otherwise.
127 """
128 if isinstance(other, LazyConfigValue):
129 other._extend.extend(self._extend)
130 self._extend = other._extend
131
132 self._prepend.extend(other._prepend)
133
134 other._inserts.extend(self._inserts)
135 self._inserts = other._inserts
136
137 if self._update:
138 other.update(self._update)
139 self._update = other._update
140 return self
141 else:
142 # other is a container, reify now.
143 return self.get_value(other)
144
145 def insert(self, index: int, other: t.Any) -> None:
146 if not isinstance(index, int):
147 raise TypeError("An integer is required")
148 self._inserts.append((index, other))
149
150 # dict methods
151 # update is used for both dict and set
152 _update = Any()
153
154 def update(self, other: t.Any) -> None:
155 """Update either a set or dict"""
156 if self._update is None:
157 if isinstance(other, dict):
158 self._update = {}
159 else:
160 self._update = set()
161 self._update.update(other)
162
163 # set methods
164 def add(self, obj: t.Any) -> None:
165 """Add an item to a set"""
166 self.update({obj})
167
168 def get_value(self, initial: t.Any) -> t.Any:
169 """construct the value from the initial one
170
171 after applying any insert / extend / update changes
172 """
173 if self._value is not None:
174 return self._value # type:ignore[unreachable]
175 value = copy.deepcopy(initial)
176 if isinstance(value, list):
177 for idx, obj in self._inserts:
178 value.insert(idx, obj)
179 value[:0] = self._prepend
180 value.extend(self._extend)
181
182 elif isinstance(value, dict):
183 if self._update:
184 value.update(self._update)
185 elif isinstance(value, set):
186 if self._update:
187 value.update(self._update)
188 self._value = value
189 return value
190
191 def to_dict(self) -> dict[str, t.Any]:
192 """return JSONable dict form of my data
193
194 Currently update as dict or set, extend, prepend as lists, and inserts as list of tuples.
195 """
196 d = {}
197 if self._update:
198 d["update"] = self._update
199 if self._extend:
200 d["extend"] = self._extend
201 if self._prepend:
202 d["prepend"] = self._prepend
203 elif self._inserts:
204 d["inserts"] = self._inserts
205 return d
206
207 def __repr__(self) -> str:
208 if self._value is not None:
209 return f"<{self.__class__.__name__} value={self._value!r}>"
210 else:
211 return f"<{self.__class__.__name__} {self.to_dict()!r}>"
212
213
214def _is_section_key(key: str) -> bool:
215 """Is a Config key a section name (does it start with a capital)?"""
216 return bool(key and key[0].upper() == key[0] and not key.startswith("_"))
217
218
219class Config(dict): # type:ignore[type-arg]
220 """An attribute-based dict that can do smart merges.
221
222 Accessing a field on a config object for the first time populates the key
223 with either a nested Config object for keys starting with capitals
224 or :class:`.LazyConfigValue` for lowercase keys,
225 allowing quick assignments such as::
226
227 c = Config()
228 c.Class.int_trait = 5
229 c.Class.list_trait.append("x")
230
231 """
232
233 def __init__(self, *args: t.Any, **kwds: t.Any) -> None:
234 dict.__init__(self, *args, **kwds)
235 self._ensure_subconfig()
236
237 def _ensure_subconfig(self) -> None:
238 """ensure that sub-dicts that should be Config objects are
239
240 casts dicts that are under section keys to Config objects,
241 which is necessary for constructing Config objects from dict literals.
242 """
243 for key in self:
244 obj = self[key]
245 if _is_section_key(key) and isinstance(obj, dict) and not isinstance(obj, Config):
246 setattr(self, key, Config(obj))
247
248 def _merge(self, other: t.Any) -> None:
249 """deprecated alias, since traitlets 4.0 - 2015 use Config.merge()"""
250 # re-added in August 2026 because of Spyder
251 import warnings
252
253 warnings.warn(
254 "_merge has been deprecated since traitlets 4.0 - 2015, please use `merge()`",
255 DeprecationWarning,
256 stacklevel=2,
257 )
258 self.merge(other)
259
260 def merge(self, other: t.Any) -> None:
261 """merge another config object into this one"""
262 to_update = {}
263 for k, v in other.items():
264 if k not in self:
265 to_update[k] = v
266 else: # I have this key
267 if isinstance(v, Config) and isinstance(self[k], Config):
268 # Recursively merge common sub Configs
269 self[k].merge(v)
270 elif isinstance(v, LazyConfigValue):
271 self[k] = v.merge_into(self[k])
272 else:
273 # Plain updates for non-Configs
274 to_update[k] = v
275
276 self.update(to_update)
277
278 def collisions(self, other: Config) -> dict[str, t.Any]:
279 """Check for collisions between two config objects.
280
281 Returns a dict of the form {"Class": {"trait": "collision message"}}`,
282 indicating which values have been ignored.
283
284 An empty dict indicates no collisions.
285 """
286 collisions: dict[str, t.Any] = {}
287 for section in self:
288 if section not in other:
289 continue
290 mine = self[section]
291 theirs = other[section]
292 for key in mine:
293 if key in theirs and mine[key] != theirs[key]:
294 collisions.setdefault(section, {})
295 collisions[section][key] = f"{mine[key]!r} ignored, using {theirs[key]!r}"
296 return collisions
297
298 def __contains__(self, key: t.Any) -> bool:
299 # allow nested contains of the form `"Section.key" in config`
300 if "." in key:
301 first, remainder = key.split(".", 1)
302 if first not in self:
303 return False
304 return remainder in self[first]
305
306 return super().__contains__(key)
307
308 # .has_key is deprecated for dictionaries.
309 has_key = __contains__
310
311 def _has_section(self, key: str) -> bool:
312 return _is_section_key(key) and key in self
313
314 def copy(self) -> dict[str, t.Any]:
315 return type(self)(dict.copy(self))
316
317 def __copy__(self) -> dict[str, t.Any]:
318 return self.copy()
319
320 def __deepcopy__(self, memo: t.Any) -> Config:
321 new_config = type(self)()
322 for key, value in self.items():
323 if isinstance(value, (Config, LazyConfigValue)):
324 # deep copy config objects
325 value = copy.deepcopy(value, memo)
326 elif type(value) in {dict, list, set, tuple}:
327 # shallow copy plain container traits
328 value = copy.copy(value)
329 new_config[key] = value
330 return new_config
331
332 def __getitem__(self, key: str) -> t.Any:
333 try:
334 return dict.__getitem__(self, key)
335 except KeyError:
336 if _is_section_key(key):
337 c = Config()
338 dict.__setitem__(self, key, c)
339 return c
340 elif not key.startswith("_"):
341 # undefined, create lazy value, used for container methods
342 v = LazyConfigValue()
343 dict.__setitem__(self, key, v)
344 return v
345 else:
346 raise
347
348 def __setitem__(self, key: str, value: t.Any) -> None:
349 if _is_section_key(key):
350 if not isinstance(value, Config):
351 raise ValueError(
352 "values whose keys begin with an uppercase "
353 f"char must be Config instances: {key!r}, {value!r}"
354 )
355 dict.__setitem__(self, key, value)
356
357 def __getattr__(self, key: str) -> t.Any:
358 if key.startswith("__"):
359 return dict.__getattr__(self, key) # type:ignore[attr-defined]
360 try:
361 return self.__getitem__(key)
362 except KeyError as e:
363 raise AttributeError(e) from e
364
365 def __setattr__(self, key: str, value: t.Any) -> None:
366 if key.startswith("__"):
367 return dict.__setattr__(self, key, value)
368 try:
369 self.__setitem__(key, value)
370 except KeyError as e:
371 raise AttributeError(e) from e
372
373 def __delattr__(self, key: str) -> None:
374 if key.startswith("__"):
375 return dict.__delattr__(self, key)
376 try:
377 dict.__delitem__(self, key)
378 except KeyError as e:
379 raise AttributeError(e) from e
380
381
382class DeferredConfig:
383 """Class for deferred-evaluation of config from CLI"""
384
385 def get_value(self, trait: TraitType[t.Any, t.Any]) -> t.Any:
386 raise NotImplementedError("Implement in subclasses")
387
388 def _super_repr(self) -> str:
389 # explicitly call super on direct parent
390 return super(self.__class__, self).__repr__()
391
392
393class DeferredConfigString(str, DeferredConfig):
394 """Config value for loading config from a string
395
396 Interpretation is deferred until it is loaded into the trait.
397
398 Subclass of str for backward compatibility.
399
400 This class is only used for values that are not listed
401 in the configurable classes.
402
403 When config is loaded, `trait.from_string` will be used.
404
405 If an error is raised in `.from_string`,
406 the original string is returned.
407
408 .. versionadded:: 5.0
409 """
410
411 def get_value(self, trait: TraitType[t.Any, t.Any]) -> t.Any:
412 """Get the value stored in this string"""
413 s = str(self)
414 try:
415 return trait.from_string(s)
416 except Exception:
417 # exception casting from string,
418 # let the original string lie.
419 # this will raise a more informative error when config is loaded.
420 return s
421
422 def __repr__(self) -> str:
423 return f"{self.__class__.__name__}({self._super_repr()})"
424
425
426class DeferredConfigList(list[t.Any], DeferredConfig):
427 """Config value for loading config from a list of strings
428
429 Interpretation is deferred until it is loaded into the trait.
430
431 This class is only used for values that are not listed
432 in the configurable classes.
433
434 When config is loaded, `trait.from_string_list` will be used.
435
436 If an error is raised in `.from_string_list`,
437 the original string list is returned.
438
439 .. versionadded:: 5.0
440 """
441
442 def get_value(self, trait: TraitType[t.Any, t.Any]) -> t.Any:
443 """Get the value stored in this string"""
444 if hasattr(trait, "from_string_list"):
445 src = list(self)
446 cast = trait.from_string_list
447 else:
448 # only allow one item
449 if len(self) > 1:
450 raise ValueError(
451 f"{trait.name} only accepts one value, got {len(self)}: {list(self)}"
452 )
453 src = self[0]
454 cast = trait.from_string
455
456 try:
457 return cast(src)
458 except Exception:
459 # exception casting from string,
460 # let the original value lie.
461 # this will raise a more informative error when config is loaded.
462 return src
463
464 def __repr__(self) -> str:
465 return f"{self.__class__.__name__}({self._super_repr()})"
466
467
468# -----------------------------------------------------------------------------
469# Config loading classes
470# -----------------------------------------------------------------------------
471
472
473class ConfigLoader:
474 """A object for loading configurations from just about anywhere.
475
476 The resulting configuration is packaged as a :class:`Config`.
477
478 Notes
479 -----
480 A :class:`ConfigLoader` does one thing: load a config from a source
481 (file, command line arguments) and returns the data as a :class:`Config` object.
482 There are lots of things that :class:`ConfigLoader` does not do. It does
483 not implement complex logic for finding config files. It does not handle
484 default values or merge multiple configs. These things need to be
485 handled elsewhere.
486 """
487
488 def _log_default(self) -> Logger:
489 from traitlets.log import get_logger
490
491 return t.cast(Logger, get_logger())
492
493 def __init__(self, log: Logger | None = None) -> None:
494 """A base class for config loaders.
495
496 log : instance of :class:`logging.Logger` to use.
497 By default logger of :meth:`traitlets.config.application.Application.instance()`
498 will be used
499
500 Examples
501 --------
502 >>> cl = ConfigLoader()
503 >>> config = cl.load_config()
504 >>> config
505 {}
506 """
507 self.clear()
508 if log is None:
509 self.log = self._log_default()
510 self.log.debug("Using default logger")
511 else:
512 self.log = log
513
514 def clear(self) -> None:
515 self.config = Config()
516
517 def load_config(self) -> Config:
518 """Load a config from somewhere, return a :class:`Config` instance.
519
520 Usually, this will cause self.config to be set and then returned.
521 However, in most cases, :meth:`ConfigLoader.clear` should be called
522 to erase any previous state.
523 """
524 self.clear()
525 return self.config
526
527
528class FileConfigLoader(ConfigLoader):
529 """A base class for file based configurations.
530
531 As we add more file based config loaders, the common logic should go
532 here.
533 """
534
535 def __init__(self, filename: str, path: str | None = None, **kw: t.Any) -> None:
536 """Build a config loader for a filename and path.
537
538 Parameters
539 ----------
540 filename : str
541 The file name of the config file.
542 path : str, list, tuple
543 The path to search for the config file on, or a sequence of
544 paths to try in order.
545 """
546 super().__init__(**kw)
547 self.filename = filename
548 self.path = path
549 self.full_filename = ""
550
551 def _find_file(self) -> None:
552 """Try to find the file by searching the paths."""
553 self.full_filename = filefind(self.filename, self.path)
554
555
556class JSONFileConfigLoader(FileConfigLoader):
557 """A JSON file loader for config
558
559 Can also act as a context manager that rewrite the configuration file to disk on exit.
560
561 Example::
562
563 with JSONFileConfigLoader('myapp.json','/home/jupyter/configurations/') as c:
564 c.MyNewConfigurable.new_value = 'Updated'
565
566 """
567
568 def load_config(self) -> Config:
569 """Load the config from a file and return it as a Config object."""
570 self.clear()
571 try:
572 self._find_file()
573 except OSError as e:
574 raise ConfigFileNotFound(str(e)) from e
575 dct = self._read_file_as_dict()
576 self.config = self._convert_to_config(dct)
577 return self.config
578
579 def _read_file_as_dict(self) -> dict[str, t.Any]:
580 with open(self.full_filename) as f:
581 return t.cast("dict[str, t.Any]", json.load(f))
582
583 def _convert_to_config(self, dictionary: dict[str, t.Any]) -> Config:
584 if "version" in dictionary:
585 version = dictionary.pop("version")
586 else:
587 version = 1
588
589 if version == 1:
590 return Config(dictionary)
591 else:
592 raise ValueError(f"Unknown version of JSON config file: {version}")
593
594 def __enter__(self) -> Config:
595 self.load_config()
596 return self.config
597
598 def __exit__(self, exc_type: object, exc_value: object, traceback: object) -> None:
599 """
600 Exit the context manager but do not handle any errors.
601
602 In case of any error, we do not want to write the potentially broken
603 configuration to disk.
604 """
605 self.config.version = 1
606 json_config = json.dumps(self.config, indent=2)
607 with open(self.full_filename, "w") as f:
608 f.write(json_config)
609
610
611class PyFileConfigLoader(FileConfigLoader):
612 """A config loader for pure python files.
613
614 This is responsible for locating a Python config file by filename and
615 path, then executing it to construct a Config object.
616 """
617
618 def load_config(self) -> Config:
619 """Load the config from a file and return it as a Config object."""
620 self.clear()
621 try:
622 self._find_file()
623 except OSError as e:
624 raise ConfigFileNotFound(str(e)) from e
625 self._read_file_as_dict()
626 return self.config
627
628 def load_subconfig(self, fname: str, path: str | None = None) -> None:
629 """Injected into config file namespace as load_subconfig"""
630 if path is None:
631 path = self.path
632
633 loader = self.__class__(fname, path)
634 try:
635 sub_config = loader.load_config()
636 except ConfigFileNotFound:
637 # Pass silently if the sub config is not there,
638 # treat it as an empty config file.
639 pass
640 else:
641 self.config.merge(sub_config)
642
643 def _read_file_as_dict(self) -> None:
644 """Load the config file into self.config, with recursive loading."""
645
646 def get_config() -> Config:
647 """Unnecessary now, but a deprecation warning is more trouble than it's worth."""
648 return self.config
649
650 namespace = dict( # noqa: C408
651 c=self.config,
652 load_subconfig=self.load_subconfig,
653 get_config=get_config,
654 __file__=self.full_filename,
655 )
656 conf_filename = self.full_filename
657 with open(conf_filename, "rb") as f:
658 exec(compile(f.read(), conf_filename, "exec"), namespace, namespace) # noqa: S102
659
660
661class CommandLineConfigLoader(ConfigLoader):
662 """A config loader for command line arguments.
663
664 As we add more command line based loaders, the common logic should go
665 here.
666 """
667
668 def _exec_config_str(
669 self, lhs: t.Any, rhs: t.Any, trait: TraitType[t.Any, t.Any] | None = None
670 ) -> None:
671 """execute self.config.<lhs> = <rhs>
672
673 * expands ~ with expanduser
674 * interprets value with trait if available
675 """
676 value = rhs
677 if isinstance(value, DeferredConfig):
678 if trait:
679 # trait available, reify config immediately
680 value = value.get_value(trait)
681 elif isinstance(rhs, DeferredConfigList) and len(rhs) == 1:
682 # single item, make it a deferred str
683 value = DeferredConfigString(os.path.expanduser(rhs[0]))
684 else:
685 if trait:
686 value = trait.from_string(value)
687 else:
688 value = DeferredConfigString(value)
689
690 *path, key = lhs.split(".")
691 section = self.config
692 for part in path:
693 section = section[part]
694 section[key] = value
695 return
696
697 def _load_flag(self, cfg: t.Any) -> None:
698 """update self.config from a flag, which can be a dict or Config"""
699 if isinstance(cfg, (dict, Config)):
700 # don't clobber whole config sections, update
701 # each section from config:
702 for sec, c in cfg.items():
703 self.config[sec].update(c)
704 else:
705 raise TypeError(f"Invalid flag: {cfg!r}")
706
707
708# match --Class.trait keys for argparse
709# matches:
710# --Class.trait
711# --x
712# -x
713
714class_trait_opt_pattern = re.compile(r"^\-?\-[A-Za-z][\w]*(\.[\w]+)*$")
715
716_DOT_REPLACEMENT = "__DOT__"
717_DASH_REPLACEMENT = "__DASH__"
718
719
720class _KVAction(argparse.Action):
721 """Custom argparse action for handling --Class.trait=x
722
723 Always
724 """
725
726 def __call__( # type:ignore[override]
727 self,
728 parser: argparse.ArgumentParser,
729 namespace: dict[str, t.Any],
730 values: t.Sequence[t.Any],
731 option_string: str | None = None,
732 ) -> None:
733 if isinstance(values, str):
734 values = [values]
735 values = ["-" if v is _DASH_REPLACEMENT else v for v in values]
736 items = getattr(namespace, self.dest, None)
737 if items is None:
738 items = DeferredConfigList()
739 else:
740 items = DeferredConfigList(items)
741 items.extend(values)
742 setattr(namespace, self.dest, items)
743
744
745class _DefaultOptionDict(dict): # type:ignore[type-arg]
746 """Like the default options dict
747
748 but acts as if all --Class.trait options are predefined
749 """
750
751 def _add_kv_action(self, key: str) -> None:
752 self[key] = _KVAction(
753 option_strings=[key],
754 dest=key.lstrip("-").replace(".", _DOT_REPLACEMENT),
755 # use metavar for display purposes
756 metavar=key.lstrip("-"),
757 )
758
759 def __contains__(self, key: t.Any) -> bool:
760 if "=" in key:
761 return False
762 if super().__contains__(key):
763 return True
764
765 if key.startswith("-") and class_trait_opt_pattern.match(key):
766 self._add_kv_action(key)
767 return True
768 return False
769
770 def __getitem__(self, key: str) -> t.Any:
771 if key in self:
772 return super().__getitem__(key)
773 else:
774 raise KeyError(key)
775
776 def get(self, key: str, default: t.Any = None) -> t.Any:
777 try:
778 return self[key]
779 except KeyError:
780 return default
781
782
783class _KVArgParser(argparse.ArgumentParser):
784 """subclass of ArgumentParser where any --Class.trait option is implicitly defined"""
785
786 def parse_known_args( # type:ignore[override]
787 self, args: t.Sequence[str] | None = None, namespace: argparse.Namespace | None = None
788 ) -> tuple[argparse.Namespace | None, list[str]]:
789 # must be done immediately prior to parsing because if we do it in init,
790 # registration of explicit actions via parser.add_option will fail during setup
791 for container in (self, self._optionals):
792 container._option_string_actions = _DefaultOptionDict(container._option_string_actions)
793 return super().parse_known_args(args, namespace)
794
795
796# type aliases
797SubcommandsDict = dict[str, t.Any]
798
799
800class ArgParseConfigLoader(CommandLineConfigLoader):
801 """A loader that uses the argparse module to load from the command line."""
802
803 parser_class = ArgumentParser
804
805 def __init__(
806 self,
807 argv: list[str] | None = None,
808 aliases: dict[str, str] | None = None,
809 flags: dict[str, str] | None = None,
810 log: t.Any = None,
811 classes: list[type[t.Any]] | None = None,
812 subcommands: SubcommandsDict | None = None,
813 *parser_args: t.Any,
814 **parser_kw: t.Any,
815 ) -> None:
816 """Create a config loader for use with argparse.
817
818 Parameters
819 ----------
820 classes : optional, list
821 The classes to scan for *container* config-traits and decide
822 for their "multiplicity" when adding them as *argparse* arguments.
823 argv : optional, list
824 If given, used to read command-line arguments from, otherwise
825 sys.argv[1:] is used.
826 *parser_args : tuple
827 A tuple of positional arguments that will be passed to the
828 constructor of :class:`argparse.ArgumentParser`.
829 **parser_kw : dict
830 A tuple of keyword arguments that will be passed to the
831 constructor of :class:`argparse.ArgumentParser`.
832 aliases : dict of str to str
833 Dict of aliases to full traitlets names for CLI parsing
834 flags : dict of str to str
835 Dict of flags to full traitlets names for CLI parsing
836 log
837 Passed to `ConfigLoader`
838 """
839 classes = classes or []
840 super(CommandLineConfigLoader, self).__init__(log=log)
841 self.clear()
842 if argv is None:
843 argv = sys.argv[1:]
844 self.argv = argv
845 self.aliases = aliases or {}
846 self.flags = flags or {}
847 self.classes = classes
848 self.subcommands = subcommands # only used for argcomplete currently
849
850 self.parser_args = parser_args
851 self.version = parser_kw.pop("version", None)
852 kwargs = dict(argument_default=argparse.SUPPRESS) # noqa: C408
853 kwargs.update(parser_kw)
854 self.parser_kw = kwargs
855
856 def load_config(
857 self,
858 argv: list[str] | None = None,
859 aliases: t.Any = None,
860 flags: t.Any = _deprecated,
861 classes: t.Any = None,
862 ) -> Config:
863 """Parse command line arguments and return as a Config object.
864
865 Parameters
866 ----------
867 argv : optional, list
868 If given, a list with the structure of sys.argv[1:] to parse
869 arguments from. If not given, the instance's self.argv attribute
870 (given at construction time) is used.
871 flags
872 Deprecated in traitlets 5.0, instantiate the config loader with the flags.
873
874 """
875
876 if flags is not _deprecated:
877 warnings.warn(
878 "The `flag` argument to load_config is deprecated since Traitlets "
879 f"5.0 and will be ignored, pass flags the `{type(self)}` constructor.",
880 DeprecationWarning,
881 stacklevel=2,
882 )
883
884 self.clear()
885 if argv is None:
886 argv = self.argv
887 if aliases is not None:
888 self.aliases = aliases
889 if classes is not None:
890 self.classes = classes
891 self._create_parser()
892 self._argcomplete(self.classes, self.subcommands)
893 self._parse_args(argv)
894 self._convert_to_config()
895 return self.config
896
897 def get_extra_args(self) -> list[str]:
898 if hasattr(self, "extra_args"):
899 return self.extra_args
900 else:
901 return []
902
903 def _create_parser(self) -> None:
904 self.parser = self.parser_class(
905 *self.parser_args,
906 **self.parser_kw, # type:ignore[arg-type]
907 )
908 self._add_arguments(self.aliases, self.flags, self.classes)
909
910 def _add_arguments(self, aliases: t.Any, flags: t.Any, classes: t.Any) -> None:
911 raise NotImplementedError("subclasses must implement _add_arguments")
912
913 def _argcomplete(self, classes: list[t.Any], subcommands: SubcommandsDict | None) -> None:
914 """If argcomplete is enabled, allow triggering command-line autocompletion"""
915
916 def _parse_args(self, args: t.Any) -> t.Any:
917 """self.parser->self.parsed_data"""
918 uargs = [cast_unicode(a) for a in args]
919
920 unpacked_aliases: dict[str, str] = {}
921 if self.aliases:
922 unpacked_aliases = {}
923 for alias, alias_target in self.aliases.items():
924 if alias in self.flags:
925 continue
926 if not isinstance(alias, tuple): # type:ignore[unreachable]
927 alias = (alias,) # type:ignore[assignment]
928 for al in alias:
929 if len(al) == 1:
930 unpacked_aliases["-" + al] = "--" + alias_target
931 unpacked_aliases["--" + al] = "--" + alias_target
932
933 def _replace(arg: str) -> str:
934 if arg == "-":
935 return _DASH_REPLACEMENT
936 for k, v in unpacked_aliases.items():
937 if arg == k:
938 return v
939 if arg.startswith(k + "="):
940 return v + "=" + arg[len(k) + 1 :]
941 return arg
942
943 if "--" in uargs:
944 idx = uargs.index("--")
945 extra_args = uargs[idx + 1 :]
946 to_parse = uargs[:idx]
947 else:
948 extra_args = []
949 to_parse = uargs
950 to_parse = [_replace(a) for a in to_parse]
951
952 self.parsed_data = self.parser.parse_args(to_parse)
953 self.extra_args = extra_args
954
955 def _convert_to_config(self) -> None:
956 """self.parsed_data->self.config"""
957 for k, v in vars(self.parsed_data).items():
958 *path, key = k.split(".")
959 section = self.config
960 for p in path:
961 section = section[p]
962 setattr(section, key, v)
963
964
965class _FlagAction(argparse.Action):
966 """ArgParse action to handle a flag"""
967
968 def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
969 self.flag = kwargs.pop("flag")
970 self.alias = kwargs.pop("alias", None)
971 kwargs["const"] = Undefined
972 if not self.alias:
973 kwargs["nargs"] = 0
974 super().__init__(*args, **kwargs)
975
976 def __call__(
977 self, parser: t.Any, namespace: t.Any, values: t.Any, option_string: str | None = None
978 ) -> None:
979 if self.nargs == 0 or values is Undefined:
980 if not hasattr(namespace, "_flags"):
981 namespace._flags = []
982 namespace._flags.append(self.flag)
983 else:
984 setattr(namespace, self.alias, values)
985
986
987class KVArgParseConfigLoader(ArgParseConfigLoader):
988 """A config loader that loads aliases and flags with argparse,
989
990 as well as arbitrary --Class.trait value
991 """
992
993 parser_class = _KVArgParser # type:ignore[assignment]
994
995 def _add_arguments(self, aliases: t.Any, flags: t.Any, classes: t.Any) -> None:
996 alias_flags: dict[str, t.Any] = {}
997 argparse_kwds: dict[str, t.Any]
998 argparse_traits: dict[str, t.Any]
999 paa = self.parser.add_argument
1000 self.parser.set_defaults(_flags=[])
1001 paa("extra_args", nargs="*")
1002
1003 # An index of all container traits collected::
1004 #
1005 # { <traitname>: (<trait>, <argparse-kwds>) }
1006 #
1007 # Used to add the correct type into the `config` tree.
1008 # Used also for aliases, not to re-collect them.
1009 self.argparse_traits = argparse_traits = {}
1010 for cls in classes:
1011 for traitname, trait in cls.class_traits(config=True).items():
1012 argname = f"{cls.__name__}.{traitname}"
1013 argparse_kwds = {"type": str}
1014 if isinstance(trait, (Container, Dict)):
1015 multiplicity = trait.metadata.get("multiplicity", "append")
1016 if multiplicity == "append":
1017 argparse_kwds["action"] = multiplicity
1018 else:
1019 argparse_kwds["nargs"] = multiplicity
1020 argparse_traits[argname] = (trait, argparse_kwds)
1021
1022 for keys, (value, fhelp) in flags.items():
1023 if not isinstance(keys, tuple):
1024 keys = (keys,)
1025 for key in keys:
1026 if key in aliases:
1027 alias_flags[aliases[key]] = value
1028 continue
1029 keys = ("-" + key, "--" + key) if len(key) == 1 else ("--" + key,)
1030 paa(*keys, action=_FlagAction, flag=value, help=fhelp)
1031
1032 for keys, traitname in aliases.items():
1033 if not isinstance(keys, tuple):
1034 keys = (keys,)
1035
1036 for key in keys:
1037 argparse_kwds = {
1038 "type": str,
1039 "dest": traitname.replace(".", _DOT_REPLACEMENT),
1040 "metavar": traitname,
1041 }
1042 argcompleter = None
1043 if traitname in argparse_traits:
1044 trait, kwds = argparse_traits[traitname]
1045 argparse_kwds.update(kwds)
1046 if "action" in argparse_kwds and traitname in alias_flags:
1047 # flag sets 'action', so can't have flag & alias with custom action
1048 # on the same name
1049 raise ArgumentError(
1050 f"The alias `{key}` for the 'append' sequence "
1051 f"config-trait `{traitname}` cannot be also a flag!'"
1052 )
1053 # For argcomplete, check if any either an argcompleter metadata tag or method
1054 # is available. If so, it should be a callable which takes the command-line key
1055 # string as an argument and other kwargs passed by argcomplete,
1056 # and returns the a list of string completions.
1057 argcompleter = trait.metadata.get("argcompleter") or getattr(
1058 trait, "argcompleter", None
1059 )
1060 if traitname in alias_flags:
1061 # alias and flag.
1062 # when called with 0 args: flag
1063 # when called with >= 1: alias
1064 argparse_kwds.setdefault("nargs", "?")
1065 argparse_kwds["action"] = _FlagAction
1066 argparse_kwds["flag"] = alias_flags[traitname]
1067 argparse_kwds["alias"] = traitname
1068 keys = ("-" + key, "--" + key) if len(key) == 1 else ("--" + key,)
1069 action = paa(*keys, **argparse_kwds)
1070 if argcompleter is not None:
1071 # argcomplete's completers are callables returning list of completion strings
1072 action.completer = functools.partial( # type:ignore[attr-defined]
1073 argcompleter, key=key
1074 )
1075
1076 def _convert_to_config(self) -> None:
1077 """self.parsed_data->self.config, parse unrecognized extra args via KVLoader."""
1078 extra_args = self.extra_args
1079
1080 for lhs, rhs in vars(self.parsed_data).items():
1081 if lhs == "extra_args":
1082 self.extra_args = ["-" if a == _DASH_REPLACEMENT else a for a in rhs] + extra_args
1083 continue
1084 if lhs == "_flags":
1085 # _flags will be handled later
1086 continue
1087
1088 lhs = lhs.replace(_DOT_REPLACEMENT, ".")
1089 if "." not in lhs:
1090 self._handle_unrecognized_alias(lhs)
1091 trait = None
1092
1093 if isinstance(rhs, list):
1094 rhs = DeferredConfigList(rhs)
1095 elif isinstance(rhs, str):
1096 rhs = DeferredConfigString(rhs)
1097
1098 trait = self.argparse_traits.get(lhs)
1099 if trait:
1100 trait = trait[0]
1101
1102 # eval the KV assignment
1103 try:
1104 self._exec_config_str(lhs, rhs, trait)
1105 except Exception as e:
1106 # cast deferred to nicer repr for the error
1107 # DeferredList->list, etc
1108 if isinstance(rhs, DeferredConfig):
1109 rhs = rhs._super_repr()
1110 raise ArgumentError(f"Error loading argument {lhs}={rhs}, {e}") from e
1111
1112 for subc in self.parsed_data._flags:
1113 self._load_flag(subc)
1114
1115 def _handle_unrecognized_alias(self, arg: str) -> None:
1116 """Handling for unrecognized alias arguments
1117
1118 Probably a mistyped alias. By default just log a warning,
1119 but users can override this to raise an error instead, e.g.
1120 self.parser.error("Unrecognized alias: '%s'" % arg)
1121 """
1122 self.log.warning("Unrecognized alias: '%s', it will have no effect.", arg)
1123
1124 def _argcomplete(self, classes: list[t.Any], subcommands: SubcommandsDict | None) -> None:
1125 """If argcomplete is enabled, allow triggering command-line autocompletion"""
1126 try:
1127 import argcomplete # noqa: F401
1128 except ImportError:
1129 return
1130
1131 from . import argcomplete_config
1132
1133 finder = argcomplete_config.ExtendedCompletionFinder() # type:ignore[no-untyped-call]
1134 finder.config_classes = classes
1135 finder.subcommands = list(subcommands or [])
1136 # for ease of testing, pass through self._argcomplete_kwargs if set
1137 finder(self.parser, **getattr(self, "_argcomplete_kwargs", {}))
1138
1139
1140class KeyValueConfigLoader(KVArgParseConfigLoader):
1141 """Deprecated in traitlets 5.0
1142
1143 Use KVArgParseConfigLoader
1144 """
1145
1146 def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
1147 warnings.warn(
1148 "KeyValueConfigLoader is deprecated since Traitlets 5.0."
1149 " Use KVArgParseConfigLoader instead.",
1150 DeprecationWarning,
1151 stacklevel=2,
1152 )
1153 super().__init__(*args, **kwargs)
1154
1155
1156def load_pyconfig_files(config_files: list[str], path: str) -> Config:
1157 """Load multiple Python config files, merging each of them in turn.
1158
1159 Parameters
1160 ----------
1161 config_files : list of str
1162 List of config files names to load and merge into the config.
1163 path : unicode
1164 The full path to the location of the config files.
1165 """
1166 config = Config()
1167 for cf in config_files:
1168 loader = PyFileConfigLoader(cf, path=path)
1169 try:
1170 next_config = loader.load_config()
1171 except ConfigFileNotFound:
1172 pass
1173 except Exception:
1174 raise
1175 else:
1176 config.merge(next_config)
1177 return config