Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.9/dist-packages/traitlets/config/configurable.py: 21%
245 statements
« prev ^ index » next coverage.py v7.3.1, created at 2023-09-25 06:05 +0000
« prev ^ index » next coverage.py v7.3.1, created at 2023-09-25 06:05 +0000
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)
114 for name in config_override_names:
115 setattr(self, name, kwargs[name])
117 # -------------------------------------------------------------------------
118 # Static trait notifiations
119 # -------------------------------------------------------------------------
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 ]
130 def _find_my_config(self, cfg):
131 """extract my config from a global Config object
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.
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::
139 [Bar, Foo.Bar, Tim.Foo.Bar]
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
155 def _load_config(self, cfg, section_names=None, traits=None):
156 """load traits from a Config object"""
158 if traits is None:
159 traits = self.traits(config=True)
160 if section_names is None:
161 section_names = self.section_names()
163 my_config = self._find_my_config(cfg)
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
184 if isinstance(self, LoggingConfigurable):
185 assert self.log is not None
186 warn = self.log.warning
187 else:
189 def warn(msg):
190 return warnings.warn(msg, UserWarning, stacklevel=9)
192 matches = get_close_matches(name, traits)
193 msg = f"Config option `{name}` not recognized by `{self.__class__.__name__}`."
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)
203 @observe("config")
204 @observe_compat
205 def _config_changed(self, change):
206 """Update all the class traits having ``config=True`` in metadata.
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)
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)
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
237 @classmethod
238 def class_get_help(cls, inst=None):
239 """Get the help string for this class in ReST format.
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)
254 @classmethod
255 def class_get_trait_help(cls, trait, inst=None, helptext=None):
256 """Get the helptext string for a single trait.
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)
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))
288 if "Enum" in trait.__class__.__name__:
289 # include Enum choices
290 lines.append(indent("Choices: %s" % trait.info()))
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))
304 return "\n".join(lines)
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))
311 @classmethod
312 def _defining_class(cls, trait, classes):
313 """Get the class that defines a trait
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
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
339 @classmethod
340 def class_config_section(cls, classes=None):
341 """Get the config section for this class.
343 Parameters
344 ----------
345 classes : list, optional
346 The list of other classes in the config file.
347 Used to reduce redundant information.
348 """
350 def c(s):
351 """return a commented, wrapped block."""
352 s = "\n\n".join(wrap_paragraphs(s, 78))
354 return "## " + s.replace("\n", "\n# ")
356 # section header
357 breaker = "#" + "-" * 78
358 parent_classes = ", ".join(p.__name__ for p in cls.__bases__ if issubclass(p, Configurable))
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("")
373 for name, trait in sorted(cls.class_traits(config=True).items()):
374 default_repr = trait.default_value_repr()
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}")
395 lines.append(f"# c.{cls.__name__}.{name} = {default_repr}")
396 lines.append("")
397 return "\n".join(lines)
399 @classmethod
400 def class_config_rst_doc(cls):
401 """Generate rST documentation for this class' config options.
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__
410 termline = classname + "." + trait.name
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)
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("")
433 help = trait.help or "No description"
434 lines.append(indent(dedent(help)))
436 # Blank line
437 lines.append("")
439 return "\n".join(lines)
442class LoggingConfigurable(Configurable):
443 """A parent class for Configurables that log.
445 Subclasses have a log trait, and the default behavior
446 is to get the logger from the currently running Application.
447 """
449 log = Any(help="Logger or LoggerAdapter instance")
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
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
470 return log.get_logger()
472 def _get_log_handler(self):
473 """Return the default Handler
475 Returns None if none can be found
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]
489class SingletonConfigurable(LoggingConfigurable):
490 """A configurable that only allows one instance.
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 """
497 _instance = None
499 @classmethod
500 def _walk_mro(cls):
501 """Walk the cls.mro() for parent classes that are also singletons
503 For use in instance()
504 """
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
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
525 @classmethod
526 def instance(cls, *args, **kwargs):
527 """Returns a global instance of this class.
529 This method create a new instance if none have previously been created
530 and returns a previously created instance is one already exists.
532 The arguments and keyword arguments passed to this method are passed
533 on to the :meth:`__init__` method of the class upon instantiation.
535 Examples
536 --------
537 Create a singleton class using instance, and retrieve it::
539 >>> from traitlets.config.configurable import SingletonConfigurable
540 >>> class Foo(SingletonConfigurable): pass
541 >>> foo = Foo.instance()
542 >>> foo == Foo.instance()
543 True
545 Create a subclass that is retrived using the base class instance::
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
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 )
569 @classmethod
570 def initialized(cls):
571 """Has an instance been created?"""
572 return hasattr(cls, "_instance") and cls._instance is not None