Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/traitlets/config/configurable.py: 22%

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

8import warnings

9from copy import deepcopy

10from textwrap import dedent

12from traitlets.traitlets import (

13 Any,

14 Container,

15 Dict,

16 HasTraits,

17 Instance,

18 default,

19 observe,

20 observe_compat,

21 validate,

22)

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):

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

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

50 def __init__(self, **kwargs):

51 """Create a configurable given a config config.

53 Parameters

54 ----------

55 config : Config

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

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

58 instance.

59 parent : Configurable instance, optional

60 The parent Configurable instance of this object.

62 Notes

63 -----

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

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

66 :func:`super`::

68 class MyConfigurable(Configurable):

69 def __init__(self, config=None):

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

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

73 This ensures that instances will be configured properly.

74 """

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

76 if parent is not None:

77 # config is implied from parent

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

79 kwargs["config"] = parent.config

80 self.parent = parent

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

84 # load kwarg traits, other than config

85 super().__init__(**kwargs)

87 # record traits set by config

88 config_override_names = set()

90 def notice_config_override(change):

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

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

94 """

95 if change.name in kwargs:

96 config_override_names.add(change.name)

98 self.observe(notice_config_override)

100 # load config

101 if config is not None:

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

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

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

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

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

107 # making that a class attribute.

108 # self.config = deepcopy(config)

109 self.config = config

110 else:

111 # allow _config_default to return something

112 self._load_config(self.config)

113 self.unobserve(notice_config_override)

114

115 for name in config_override_names:

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

117

118 # -------------------------------------------------------------------------

119 # Static trait notifiations

120 # -------------------------------------------------------------------------

121

122 @classmethod

123 def section_names(cls):

124 """return section names as a list"""

125 return [

126 c.__name__

127 for c in reversed(cls.__mro__)

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

129 ]

130

131 def _find_my_config(self, cfg):

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

133

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

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

136

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

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

139

140 [Bar, Foo.Bar, Tim.Foo.Bar]

141

142 With the last item being the highest priority.

143 """

144 cfgs = [cfg]

145 if self.parent:

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

147 my_config = Config()

148 for c in cfgs:

149 for sname in self.section_names():

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

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

152 if c._has_section(sname):

153 my_config.merge(c[sname])

154 return my_config

155

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

157 """load traits from a Config object"""

158

159 if traits is None:

160 traits = self.traits(config=True)

161 if section_names is None:

162 section_names = self.section_names()

163

164 my_config = self._find_my_config(cfg)

165

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

167 with self.hold_trait_notifications():

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

169 if name in traits:

170 if isinstance(config_value, LazyConfigValue):

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

172 # without having to copy the initial value

173 initial = getattr(self, name)

174 config_value = config_value.get_value(initial)

175 elif isinstance(config_value, DeferredConfig):

176 # DeferredConfig tends to come from CLI/environment variables

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

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

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

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

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

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

183 from difflib import get_close_matches

184

185 if isinstance(self, LoggingConfigurable):

186 warn = self.log.warning

187 else:

188 warn = lambda msg: warnings.warn(msg, stacklevel=9) # noqa[E371]

189 matches = get_close_matches(name, traits)

190 msg = "Config option `{option}` not recognized by `{klass}`.".format(

191 option=name, klass=self.__class__.__name__

192 )

193

194 if len(matches) == 1:

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

196 elif len(matches) >= 1:

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

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

199 )

200 warn(msg)

201

202 @observe("config")

203 @observe_compat

204 def _config_changed(self, change):

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

206

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

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

209 config entry.

210 """

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

212 traits = self.traits(config=True)

213

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

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

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

217 section_names = self.section_names()

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

219

220 def update_config(self, config):

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

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

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

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

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

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

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

228 self.config = deepcopy(self.config)

229 # load config

230 self._load_config(config)

231 # merge it into self.config

232 self.config.merge(config)

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

234 # DO NOT trigger full trait-change

235

236 @classmethod

237 def class_get_help(cls, inst=None):

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

239

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

241 class defaults.

242 """

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

244 final_help = []

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

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

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

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

249 help = cls.class_get_trait_help(v, inst)

250 final_help.append(help)

251 return "\n".join(final_help)

252

253 @classmethod

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

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

256

257 :param inst:

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

259 the class default.

260 :param helptext:

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

262 """

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

264 lines = []

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

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

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

268 if isinstance(trait, Dict):

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

270 else:

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

272 if multiplicity == "append":

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

274 else:

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

276 else:

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

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

279 lines.append(header)

280

281 if helptext is None:

282 helptext = trait.help

283 if helptext != "":

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

285 lines.append(indent(helptext))

286

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

288 # include Enum choices

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

290

291 if inst is not None:

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

293 else:

294 try:

295 dvr = trait.default_value_repr()

296 except Exception:

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

298 if dvr is not None:

299 if len(dvr) > 64:

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

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

302

303 return "\n".join(lines)

304

305 @classmethod

306 def class_print_help(cls, inst=None):

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

308 print(cls.class_get_help(inst))

309

310 @classmethod

311 def _defining_class(cls, trait, classes):

312 """Get the class that defines a trait

313

314 For reducing redundant help output in config files.

315 Returns the current class if:

316 - the trait is defined on this class, or

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

318

319 Parameters

320 ----------

321 trait : Trait

322 The trait to look for

323 classes : list

324 The list of other classes to consider for redundancy.

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

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

327 """

328 defining_cls = cls

329 for parent in cls.mro():

330 if (

331 issubclass(parent, Configurable)

332 and parent in classes

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

334 ):

335 defining_cls = parent

336 return defining_cls

337

338 @classmethod

339 def class_config_section(cls, classes=None):

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

341

342 Parameters

343 ----------

344 classes : list, optional

345 The list of other classes in the config file.

346 Used to reduce redundant information.

347 """

348

349 def c(s):

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

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

352

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

354

355 # section header

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

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

358

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

360 lines = [breaker, s, breaker]

361 # get the description trait

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

363 if desc:

364 desc = desc.default_value

365 if not desc:

366 # no description from trait, use __doc__

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

368 if desc:

369 lines.append(c(desc))

370 lines.append("")

371

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

373 default_repr = trait.default_value_repr()

374

375 if classes:

376 defining_class = cls._defining_class(trait, classes)

377 else:

378 defining_class = cls

379 if defining_class is cls:

380 # cls owns the trait, show full help

381 if trait.help:

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

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

384 # include Enum choices

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

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

387 else:

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

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

390 if trait.help:

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

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

393

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

395 lines.append("")

396 return "\n".join(lines)

397

398 @classmethod

399 def class_config_rst_doc(cls):

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

401

402 Excludes traits defined on parent classes.

403 """

404 lines = []

405 classname = cls.__name__

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

407 ttype = trait.__class__.__name__

408

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

410

411 # Choices or type

412 if "Enum" in ttype:

413 # include Enum choices

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

415 else:

416 termline += " : " + ttype

417 lines.append(termline)

418

419 # Default value

420 try:

421 dvr = trait.default_value_repr()

422 except Exception:

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

424 if dvr is not None:

425 if len(dvr) > 64:

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

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

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

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

430 lines.append("")

431

432 help = trait.help or "No description"

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

434

435 # Blank line

436 lines.append("")

437

438 return "\n".join(lines)

439

440

441class LoggingConfigurable(Configurable):

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

443

444 Subclasses have a log trait, and the default behavior

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

446 """

447

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

449

450 @validate("log")

451 def _validate_log(self, proposal):

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

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

454 warnings.warn(

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

456 f" got {proposal.value}."

457 )

458 return proposal.value

459

460 @default("log")

461 def _log_default(self):

462 if isinstance(self.parent, LoggingConfigurable):

463 return self.parent.log

464 from traitlets import log

465

466 return log.get_logger()

467

468 def _get_log_handler(self):

469 """Return the default Handler

470

471 Returns None if none can be found

472

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

474 not be the default one.

475 """

476 logger = self.log

477 if isinstance(logger, logging.LoggerAdapter):

478 logger = logger.logger

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

480 # no handlers attribute or empty handlers list

481 return None

482 return logger.handlers[0]

483

484

485class SingletonConfigurable(LoggingConfigurable):

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

487

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

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

490 :meth:`SingletonConfigurable.instance` method.

491 """

492

493 _instance = None

494

495 @classmethod

496 def _walk_mro(cls):

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

498

499 For use in instance()

500 """

501

502 for subclass in cls.mro():

503 if (

504 issubclass(cls, subclass)

505 and issubclass(subclass, SingletonConfigurable)

506 and subclass != SingletonConfigurable

507 ):

508 yield subclass

509

510 @classmethod

511 def clear_instance(cls):

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

513 if not cls.initialized():

514 return

515 for subclass in cls._walk_mro():

516 if isinstance(subclass._instance, cls):

517 # only clear instances that are instances

518 # of the calling class

519 subclass._instance = None

520

521 @classmethod

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

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

524

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

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

527

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

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

530

531 Examples

532 --------

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

534

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

536 >>> class Foo(SingletonConfigurable): pass

537 >>> foo = Foo.instance()

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

539 True

540

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

542

543 >>> class Bar(SingletonConfigurable): pass

544 >>> class Bam(Bar): pass

545 >>> bam = Bam.instance()

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

547 True

548 """

549 # Create and save the instance

550 if cls._instance is None:

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

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

553 # parent classes' _instance attribute.

554 for subclass in cls._walk_mro():

555 subclass._instance = inst

556

557 if isinstance(cls._instance, cls):

558 return cls._instance

559 else:

560 raise MultipleInstanceError(

561 "An incompatible sibling of '%s' is already instantiated"

562 " as singleton: %s" % (cls.__name__, type(cls._instance).__name__)

563 )

564

565 @classmethod

566 def initialized(cls):

567 """Has an instance been created?"""

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