Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/traitlets/config/loader.py: 26%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

568 statements  

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