Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.9/dist-packages/traitlets/config/configurable.py: 21%

1"""A base class for objects that are configurable."""

3# Copyright (c) IPython Development Team.

4# Distributed under the terms of the Modified BSD License.

7import logging

8from copy import deepcopy

9from textwrap import dedent

11from traitlets.traitlets import (

12 Any,

13 Container,

14 Dict,

15 HasTraits,

16 Instance,

17 default,

18 observe,

19 observe_compat,

20 validate,

21)

22from traitlets.utils import warnings

23from traitlets.utils.text import indent, wrap_paragraphs

25from .loader import Config, DeferredConfig, LazyConfigValue, _is_section_key

27# -----------------------------------------------------------------------------

28# Helper classes for Configurables

29# -----------------------------------------------------------------------------

32class ConfigurableError(Exception):

33 pass

36class MultipleInstanceError(ConfigurableError):

37 pass

40# -----------------------------------------------------------------------------

41# Configurable implementation

42# -----------------------------------------------------------------------------

45class Configurable(HasTraits):

46 config = Instance(Config, (), {})

47 parent = Instance("traitlets.config.configurable.Configurable", allow_none=True)

49 def __init__(self, **kwargs):

50 """Create a configurable given a config config.

52 Parameters

53 ----------

54 config : Config

55 If this is empty, default values are used. If config is a

56 :class:`Config` instance, it will be used to configure the

57 instance.

58 parent : Configurable instance, optional

59 The parent Configurable instance of this object.

61 Notes

62 -----

63 Subclasses of Configurable must call the :meth:`__init__` method of

64 :class:`Configurable` *before* doing anything else and using

65 :func:`super`::

67 class MyConfigurable(Configurable):

68 def __init__(self, config=None):

69 super(MyConfigurable, self).__init__(config=config)

70 # Then any other code you need to finish initialization.

72 This ensures that instances will be configured properly.

73 """

74 parent = kwargs.pop("parent", None)

75 if parent is not None:

76 # config is implied from parent

77 if kwargs.get("config", None) is None:

78 kwargs["config"] = parent.config

79 self.parent = parent

81 config = kwargs.pop("config", None)

83 # load kwarg traits, other than config

84 super().__init__(**kwargs)

86 # record traits set by config

87 config_override_names = set()

89 def notice_config_override(change):

90 """Record traits set by both config and kwargs.

92 They will need to be overridden again after loading config.

93 """

94 if change.name in kwargs:

95 config_override_names.add(change.name)

97 self.observe(notice_config_override)

99 # load config

100 if config is not None:

101 # We used to deepcopy, but for now we are trying to just save

102 # by reference. This *could* have side effects as all components

103 # will share config. In fact, I did find such a side effect in

104 # _config_changed below. If a config attribute value was a mutable type

105 # all instances of a component were getting the same copy, effectively

106 # making that a class attribute.

107 # self.config = deepcopy(config)

108 self.config = config

109 else:

110 # allow _config_default to return something

111 self._load_config(self.config)

112 self.unobserve(notice_config_override)

113

114 for name in config_override_names:

115 setattr(self, name, kwargs[name])

116

117 # -------------------------------------------------------------------------

118 # Static trait notifiations

119 # -------------------------------------------------------------------------

120

121 @classmethod

122 def section_names(cls):

123 """return section names as a list"""

124 return [

125 c.__name__

126 for c in reversed(cls.__mro__)

127 if issubclass(c, Configurable) and issubclass(cls, c)

128 ]

129

130 def _find_my_config(self, cfg):

131 """extract my config from a global Config object

132

133 will construct a Config object of only the config values that apply to me

134 based on my mro(), as well as those of my parent(s) if they exist.

135

136 If I am Bar and my parent is Foo, and their parent is Tim,

137 this will return merge following config sections, in this order::

138

139 [Bar, Foo.Bar, Tim.Foo.Bar]

140

141 With the last item being the highest priority.

142 """

143 cfgs = [cfg]

144 if self.parent:

145 cfgs.append(self.parent._find_my_config(cfg))

146 my_config = Config()

147 for c in cfgs:

148 for sname in self.section_names():

149 # Don't do a blind getattr as that would cause the config to

150 # dynamically create the section with name Class.__name__.

151 if c._has_section(sname):

152 my_config.merge(c[sname])

153 return my_config

154

155 def _load_config(self, cfg, section_names=None, traits=None):

156 """load traits from a Config object"""

157

158 if traits is None:

159 traits = self.traits(config=True)

160 if section_names is None:

161 section_names = self.section_names()

162

163 my_config = self._find_my_config(cfg)

164

165 # hold trait notifications until after all config has been loaded

166 with self.hold_trait_notifications():

167 for name, config_value in my_config.items():

168 if name in traits:

169 if isinstance(config_value, LazyConfigValue):

170 # ConfigValue is a wrapper for using append / update on containers

171 # without having to copy the initial value

172 initial = getattr(self, name)

173 config_value = config_value.get_value(initial)

174 elif isinstance(config_value, DeferredConfig):

175 # DeferredConfig tends to come from CLI/environment variables

176 config_value = config_value.get_value(traits[name])

177 # We have to do a deepcopy here if we don't deepcopy the entire

178 # config object. If we don't, a mutable config_value will be

179 # shared by all instances, effectively making it a class attribute.

180 setattr(self, name, deepcopy(config_value))

181 elif not _is_section_key(name) and not isinstance(config_value, Config):

182 from difflib import get_close_matches

183

184 if isinstance(self, LoggingConfigurable):

185 assert self.log is not None

186 warn = self.log.warning

187 else:

188

189 def warn(msg):

190 return warnings.warn(msg, UserWarning, stacklevel=9)

191

192 matches = get_close_matches(name, traits)

193 msg = f"Config option `{name}` not recognized by `{self.__class__.__name__}`."

194

195 if len(matches) == 1:

196 msg += f" Did you mean `{matches[0]}`?"

197 elif len(matches) >= 1:

198 msg += " Did you mean one of: `{matches}`?".format(

199 matches=", ".join(sorted(matches))

200 )

201 warn(msg)

202

203 @observe("config")

204 @observe_compat

205 def _config_changed(self, change):

206 """Update all the class traits having ``config=True`` in metadata.

207

208 For any class trait with a ``config`` metadata attribute that is

209 ``True``, we update the trait with the value of the corresponding

210 config entry.

211 """

212 # Get all traits with a config metadata entry that is True

213 traits = self.traits(config=True)

214

215 # We auto-load config section for this class as well as any parent

216 # classes that are Configurable subclasses. This starts with Configurable

217 # and works down the mro loading the config for each section.

218 section_names = self.section_names()

219 self._load_config(change.new, traits=traits, section_names=section_names)

220

221 def update_config(self, config):

222 """Update config and load the new values"""

223 # traitlets prior to 4.2 created a copy of self.config in order to trigger change events.

224 # Some projects (IPython < 5) relied upon one side effect of this,

225 # that self.config prior to update_config was not modified in-place.

226 # For backward-compatibility, we must ensure that self.config

227 # is a new object and not modified in-place,

228 # but config consumers should not rely on this behavior.

229 self.config = deepcopy(self.config)

230 # load config

231 self._load_config(config)

232 # merge it into self.config

233 self.config.merge(config)

234 # TODO: trigger change event if/when dict-update change events take place

235 # DO NOT trigger full trait-change

236

237 @classmethod

238 def class_get_help(cls, inst=None):

239 """Get the help string for this class in ReST format.

240

241 If `inst` is given, its current trait values will be used in place of

242 class defaults.

243 """

244 assert inst is None or isinstance(inst, cls)

245 final_help = []

246 base_classes = ", ".join(p.__name__ for p in cls.__bases__)

247 final_help.append(f"{cls.__name__}({base_classes}) options")

248 final_help.append(len(final_help[0]) * "-")

249 for _, v in sorted(cls.class_traits(config=True).items()):

250 help = cls.class_get_trait_help(v, inst)

251 final_help.append(help)

252 return "\n".join(final_help)

253

254 @classmethod

255 def class_get_trait_help(cls, trait, inst=None, helptext=None):

256 """Get the helptext string for a single trait.

257

258 :param inst:

259 If given, its current trait values will be used in place of

260 the class default.

261 :param helptext:

262 If not given, uses the `help` attribute of the current trait.

263 """

264 assert inst is None or isinstance(inst, cls)

265 lines = []

266 header = f"--{cls.__name__}.{trait.name}"

267 if isinstance(trait, (Container, Dict)):

268 multiplicity = trait.metadata.get("multiplicity", "append")

269 if isinstance(trait, Dict):

270 sample_value = "<key-1>=<value-1>"

271 else:

272 sample_value = "<%s-item-1>" % trait.__class__.__name__.lower()

273 if multiplicity == "append":

274 header = f"{header}={sample_value}..."

275 else:

276 header = f"{header} {sample_value}..."

277 else:

278 header = f"{header}=<{trait.__class__.__name__}>"

279 # header = "--%s.%s=<%s>" % (cls.__name__, trait.name, trait.__class__.__name__)

280 lines.append(header)

281

282 if helptext is None:

283 helptext = trait.help

284 if helptext != "":

285 helptext = "\n".join(wrap_paragraphs(helptext, 76))

286 lines.append(indent(helptext))

287

288 if "Enum" in trait.__class__.__name__:

289 # include Enum choices

290 lines.append(indent("Choices: %s" % trait.info()))

291

292 if inst is not None:

293 lines.append(indent(f"Current: {getattr(inst, trait.name)!r}"))

294 else:

295 try:

296 dvr = trait.default_value_repr()

297 except Exception:

298 dvr = None # ignore defaults we can't construct

299 if dvr is not None:

300 if len(dvr) > 64:

301 dvr = dvr[:61] + "..."

302 lines.append(indent("Default: %s" % dvr))

303

304 return "\n".join(lines)

305

306 @classmethod

307 def class_print_help(cls, inst=None):

308 """Get the help string for a single trait and print it."""

309 print(cls.class_get_help(inst))

310

311 @classmethod

312 def _defining_class(cls, trait, classes):

313 """Get the class that defines a trait

314

315 For reducing redundant help output in config files.

316 Returns the current class if:

317 - the trait is defined on this class, or

318 - the class where it is defined would not be in the config file

319

320 Parameters

321 ----------

322 trait : Trait

323 The trait to look for

324 classes : list

325 The list of other classes to consider for redundancy.

326 Will return `cls` even if it is not defined on `cls`

327 if the defining class is not in `classes`.

328 """

329 defining_cls = cls

330 for parent in cls.mro():

331 if (

332 issubclass(parent, Configurable)

333 and parent in classes

334 and parent.class_own_traits(config=True).get(trait.name, None) is trait

335 ):

336 defining_cls = parent

337 return defining_cls

338

339 @classmethod

340 def class_config_section(cls, classes=None):

341 """Get the config section for this class.

342

343 Parameters

344 ----------

345 classes : list, optional

346 The list of other classes in the config file.

347 Used to reduce redundant information.

348 """

349

350 def c(s):

351 """return a commented, wrapped block."""

352 s = "\n\n".join(wrap_paragraphs(s, 78))

353

354 return "## " + s.replace("\n", "\n# ")

355

356 # section header

357 breaker = "#" + "-" * 78

358 parent_classes = ", ".join(p.__name__ for p in cls.__bases__ if issubclass(p, Configurable))

359

360 s = f"# {cls.__name__}({parent_classes}) configuration"

361 lines = [breaker, s, breaker]

362 # get the description trait

363 desc = cls.class_traits().get("description")

364 if desc:

365 desc = desc.default_value

366 if not desc:

367 # no description from trait, use __doc__

368 desc = getattr(cls, "__doc__", "")

369 if desc:

370 lines.append(c(desc))

371 lines.append("")

372

373 for name, trait in sorted(cls.class_traits(config=True).items()):

374 default_repr = trait.default_value_repr()

375

376 if classes:

377 defining_class = cls._defining_class(trait, classes)

378 else:

379 defining_class = cls

380 if defining_class is cls:

381 # cls owns the trait, show full help

382 if trait.help:

383 lines.append(c(trait.help))

384 if "Enum" in type(trait).__name__:

385 # include Enum choices

386 lines.append("# Choices: %s" % trait.info())

387 lines.append("# Default: %s" % default_repr)

388 else:

389 # Trait appears multiple times and isn't defined here.

390 # Truncate help to first line + "See also Original.trait"

391 if trait.help:

392 lines.append(c(trait.help.split("\n", 1)[0]))

393 lines.append(f"# See also: {defining_class.__name__}.{name}")

394

395 lines.append(f"# c.{cls.__name__}.{name} = {default_repr}")

396 lines.append("")

397 return "\n".join(lines)

398

399 @classmethod

400 def class_config_rst_doc(cls):

401 """Generate rST documentation for this class' config options.

402

403 Excludes traits defined on parent classes.

404 """

405 lines = []

406 classname = cls.__name__

407 for _, trait in sorted(cls.class_traits(config=True).items()):

408 ttype = trait.__class__.__name__

409

410 termline = classname + "." + trait.name

411

412 # Choices or type

413 if "Enum" in ttype:

414 # include Enum choices

415 termline += " : " + trait.info_rst()

416 else:

417 termline += " : " + ttype

418 lines.append(termline)

419

420 # Default value

421 try:

422 dvr = trait.default_value_repr()

423 except Exception:

424 dvr = None # ignore defaults we can't construct

425 if dvr is not None:

426 if len(dvr) > 64:

427 dvr = dvr[:61] + "..."

428 # Double up backslashes, so they get to the rendered docs

429 dvr = dvr.replace("\\n", "\\\\n")

430 lines.append(indent("Default: ``%s``" % dvr))

431 lines.append("")

432

433 help = trait.help or "No description"

434 lines.append(indent(dedent(help)))

435

436 # Blank line

437 lines.append("")

438

439 return "\n".join(lines)

440

441

442class LoggingConfigurable(Configurable):

443 """A parent class for Configurables that log.

444

445 Subclasses have a log trait, and the default behavior

446 is to get the logger from the currently running Application.

447 """

448

449 log = Any(help="Logger or LoggerAdapter instance")

450

451 @validate("log")

452 def _validate_log(self, proposal):

453 if not isinstance(proposal.value, (logging.Logger, logging.LoggerAdapter)):

454 # warn about unsupported type, but be lenient to allow for duck typing

455 warnings.warn(

456 f"{self.__class__.__name__}.log should be a Logger or LoggerAdapter,"

457 f" got {proposal.value}.",

458 UserWarning,

459 stacklevel=2,

460 )

461 return proposal.value

462

463 @default("log")

464 def _log_default(self):

465 if isinstance(self.parent, LoggingConfigurable):

466 assert self.parent is not None

467 return self.parent.log

468 from traitlets import log

469

470 return log.get_logger()

471

472 def _get_log_handler(self):

473 """Return the default Handler

474

475 Returns None if none can be found

476

477 Deprecated, this now returns the first log handler which may or may

478 not be the default one.

479 """

480 logger = self.log

481 if isinstance(logger, logging.LoggerAdapter):

482 logger = logger.logger

483 if not getattr(logger, "handlers", None):

484 # no handlers attribute or empty handlers list

485 return None

486 return logger.handlers[0]

487

488

489class SingletonConfigurable(LoggingConfigurable):

490 """A configurable that only allows one instance.

491

492 This class is for classes that should only have one instance of itself

493 or *any* subclass. To create and retrieve such a class use the

494 :meth:`SingletonConfigurable.instance` method.

495 """

496

497 _instance = None

498

499 @classmethod

500 def _walk_mro(cls):

501 """Walk the cls.mro() for parent classes that are also singletons

502

503 For use in instance()

504 """

505

506 for subclass in cls.mro():

507 if (

508 issubclass(cls, subclass)

509 and issubclass(subclass, SingletonConfigurable)

510 and subclass != SingletonConfigurable

511 ):

512 yield subclass

513

514 @classmethod

515 def clear_instance(cls):

516 """unset _instance for this class and singleton parents."""

517 if not cls.initialized():

518 return

519 for subclass in cls._walk_mro():

520 if isinstance(subclass._instance, cls):

521 # only clear instances that are instances

522 # of the calling class

523 subclass._instance = None

524

525 @classmethod

526 def instance(cls, *args, **kwargs):

527 """Returns a global instance of this class.

528

529 This method create a new instance if none have previously been created

530 and returns a previously created instance is one already exists.

531

532 The arguments and keyword arguments passed to this method are passed

533 on to the :meth:`__init__` method of the class upon instantiation.

534

535 Examples

536 --------

537 Create a singleton class using instance, and retrieve it::

538

539 >>> from traitlets.config.configurable import SingletonConfigurable

540 >>> class Foo(SingletonConfigurable): pass

541 >>> foo = Foo.instance()

542 >>> foo == Foo.instance()

543 True

544

545 Create a subclass that is retrived using the base class instance::

546

547 >>> class Bar(SingletonConfigurable): pass

548 >>> class Bam(Bar): pass

549 >>> bam = Bam.instance()

550 >>> bam == Bar.instance()

551 True

552 """

553 # Create and save the instance

554 if cls._instance is None:

555 inst = cls(*args, **kwargs)

556 # Now make sure that the instance will also be returned by

557 # parent classes' _instance attribute.

558 for subclass in cls._walk_mro():

559 subclass._instance = inst

560

561 if isinstance(cls._instance, cls):

562 return cls._instance

563 else:

564 raise MultipleInstanceError(

565 f"An incompatible sibling of '{cls.__name__}' is already instantiated"

566 f" as singleton: {type(cls._instance).__name__}"

567 )

568

569 @classmethod

570 def initialized(cls):

571 """Has an instance been created?"""

572 return hasattr(cls, "_instance") and cls._instance is not None