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
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.
5from __future__ import annotations
7import logging
8import typing as t
9from copy import deepcopy
10from textwrap import dedent
12from traitlets.traitlets import (
13 Any,
14 Container,
15 Dict,
16 HasTraits,
17 Instance,
18 TraitType,
19 default,
20 observe,
21 observe_compat,
22 validate,
23)
24from traitlets.utils import warnings
25from traitlets.utils.bunch import Bunch
26from traitlets.utils.text import indent, wrap_paragraphs
28from .loader import Config, DeferredConfig, LazyConfigValue, _is_section_key
30if t.TYPE_CHECKING:
31 from typing_extensions import Self
33# -----------------------------------------------------------------------------
34# Helper classes for Configurables
35# -----------------------------------------------------------------------------
37if t.TYPE_CHECKING:
38 LoggerType = logging.Logger | logging.LoggerAdapter[t.Any]
39else:
40 LoggerType = t.Any
43class ConfigurableError(Exception):
44 pass
47class MultipleInstanceError(ConfigurableError):
48 pass
51# -----------------------------------------------------------------------------
52# Configurable implementation
53# -----------------------------------------------------------------------------
56class Configurable(HasTraits):
57 config = Instance(Config, (), {})
58 parent = Instance("traitlets.config.configurable.Configurable", allow_none=True)
60 def __init__(self, **kwargs: t.Any) -> None:
61 """Create a configurable given a config config.
63 Parameters
64 ----------
65 config : Config
66 If this is empty, default values are used. If config is a
67 :class:`Config` instance, it will be used to configure the
68 instance.
69 parent : Configurable instance, optional
70 The parent Configurable instance of this object.
72 Notes
73 -----
74 Subclasses of Configurable must call the :meth:`__init__` method of
75 :class:`Configurable` *before* doing anything else and using
76 :func:`super`::
78 class MyConfigurable(Configurable):
79 def __init__(self, config=None):
80 super(MyConfigurable, self).__init__(config=config)
81 # Then any other code you need to finish initialization.
83 This ensures that instances will be configured properly.
84 """
85 parent = kwargs.pop("parent", None)
86 if parent is not None:
87 # config is implied from parent
88 if kwargs.get("config") is None:
89 kwargs["config"] = parent.config
90 self.parent = parent
92 config = kwargs.pop("config", None)
94 # load kwarg traits, other than config
95 super().__init__(**kwargs)
97 # record traits set by config
98 config_override_names = set()
100 def notice_config_override(change: Bunch) -> None:
101 """Record traits set by both config and kwargs.
103 They will need to be overridden again after loading config.
104 """
105 if change.name in kwargs:
106 config_override_names.add(change.name)
108 self.observe(notice_config_override)
110 # load config
111 if config is not None:
112 # We used to deepcopy, but for now we are trying to just save
113 # by reference. This *could* have side effects as all components
114 # will share config. In fact, I did find such a side effect in
115 # _config_changed below. If a config attribute value was a mutable type
116 # all instances of a component were getting the same copy, effectively
117 # making that a class attribute.
118 # self.config = deepcopy(config)
119 self.config = config
120 else:
121 # allow _config_default to return something
122 self._load_config(self.config)
123 self.unobserve(notice_config_override)
125 for name in config_override_names:
126 setattr(self, name, kwargs[name])
128 # -------------------------------------------------------------------------
129 # Static trait notifications
130 # -------------------------------------------------------------------------
132 @classmethod
133 def section_names(cls) -> list[str]:
134 """return section names as a list"""
135 return [
136 c.__name__
137 for c in reversed(cls.__mro__)
138 if issubclass(c, Configurable) and issubclass(cls, c)
139 ]
141 def _find_my_config(self, cfg: Config) -> t.Any:
142 """extract my config from a global Config object
144 will construct a Config object of only the config values that apply to me
145 based on my mro(), as well as those of my parent(s) if they exist.
147 If I am Bar and my parent is Foo, and their parent is Tim,
148 this will return merge following config sections, in this order::
150 [Bar, Foo.Bar, Tim.Foo.Bar]
152 With the last item being the highest priority.
153 """
154 cfgs = [cfg]
155 if self.parent:
156 cfgs.append(self.parent._find_my_config(cfg))
157 my_config = Config()
158 for c in cfgs:
159 for sname in self.section_names():
160 # Don't do a blind getattr as that would cause the config to
161 # dynamically create the section with name Class.__name__.
162 if c._has_section(sname):
163 my_config.merge(c[sname])
164 return my_config
166 def _load_config(
167 self,
168 cfg: Config,
169 section_names: list[str] | None = None,
170 traits: dict[str, TraitType[t.Any, t.Any]] | None = None,
171 ) -> None:
172 """load traits from a Config object"""
174 if traits is None:
175 traits = self.traits(config=True)
176 if section_names is None:
177 section_names = self.section_names()
179 my_config = self._find_my_config(cfg)
181 # hold trait notifications until after all config has been loaded
182 with self.hold_trait_notifications():
183 for name, config_value in my_config.items():
184 if name in traits:
185 if isinstance(config_value, LazyConfigValue):
186 # ConfigValue is a wrapper for using append / update on containers
187 # without having to copy the initial value
188 initial = getattr(self, name)
189 config_value = config_value.get_value(initial)
190 elif isinstance(config_value, DeferredConfig):
191 # DeferredConfig tends to come from CLI/environment variables
192 config_value = config_value.get_value(traits[name])
193 # We have to do a deepcopy here if we don't deepcopy the entire
194 # config object. If we don't, a mutable config_value will be
195 # shared by all instances, effectively making it a class attribute.
196 setattr(self, name, deepcopy(config_value))
197 elif not _is_section_key(name) and not isinstance(config_value, Config):
198 from difflib import get_close_matches
200 if isinstance(self, LoggingConfigurable):
201 assert self.log is not None
202 warn = self.log.warning
203 else:
205 def warn(msg: t.Any) -> None:
206 return warnings.warn(msg, UserWarning, stacklevel=9)
208 matches = get_close_matches(name, traits)
209 msg = f"Config option `{name}` not recognized by `{self.__class__.__name__}`."
211 if len(matches) == 1:
212 msg += f" Did you mean `{matches[0]}`?"
213 elif len(matches) >= 1:
214 msg += " Did you mean one of: `{matches}`?".format(
215 matches=", ".join(sorted(matches))
216 )
217 warn(msg)
219 @observe("config")
220 @observe_compat
221 def _config_changed(self, change: Bunch) -> None:
222 """Update all the class traits having ``config=True`` in metadata.
224 For any class trait with a ``config`` metadata attribute that is
225 ``True``, we update the trait with the value of the corresponding
226 config entry.
227 """
228 # Get all traits with a config metadata entry that is True
229 traits = self.traits(config=True)
231 # We auto-load config section for this class as well as any parent
232 # classes that are Configurable subclasses. This starts with Configurable
233 # and works down the mro loading the config for each section.
234 section_names = self.section_names()
235 self._load_config(change.new, traits=traits, section_names=section_names)
237 def update_config(self, config: Config) -> None:
238 """Update config and load the new values"""
239 # traitlets prior to 4.2 created a copy of self.config in order to trigger change events.
240 # Some projects (IPython < 5) relied upon one side effect of this,
241 # that self.config prior to update_config was not modified in-place.
242 # For backward-compatibility, we must ensure that self.config
243 # is a new object and not modified in-place,
244 # but config consumers should not rely on this behavior.
245 self.config = deepcopy(self.config)
246 # load config
247 self._load_config(config)
248 # merge it into self.config
249 self.config.merge(config)
250 # TODO: trigger change event if/when dict-update change events take place
251 # DO NOT trigger full trait-change
253 @classmethod
254 def class_get_help(cls, inst: HasTraits | None = None) -> str:
255 """Get the help string for this class in ReST format.
257 If `inst` is given, its current trait values will be used in place of
258 class defaults.
259 """
260 assert inst is None or isinstance(inst, cls)
261 final_help = []
262 base_classes = ", ".join(p.__name__ for p in cls.__bases__)
263 final_help.append(f"{cls.__name__}({base_classes}) options")
264 final_help.append(len(final_help[0]) * "-")
265 for _, v in sorted(cls.class_traits(config=True).items()):
266 help = cls.class_get_trait_help(v, inst)
267 final_help.append(help)
268 return "\n".join(final_help)
270 @classmethod
271 def class_get_trait_help(
272 cls,
273 trait: TraitType[t.Any, t.Any],
274 inst: HasTraits | None = None,
275 helptext: str | None = None,
276 ) -> str:
277 """Get the helptext string for a single trait.
279 :param inst:
280 If given, its current trait values will be used in place of
281 the class default.
282 :param helptext:
283 If not given, uses the `help` attribute of the current trait.
284 """
285 assert inst is None or isinstance(inst, cls)
286 lines = []
287 header = f"--{cls.__name__}.{trait.name}"
288 if isinstance(trait, (Container, Dict)):
289 multiplicity = trait.metadata.get("multiplicity", "append")
290 if isinstance(trait, Dict):
291 sample_value = "<key-1>=<value-1>"
292 else:
293 sample_value = f"<{trait.__class__.__name__.lower()}-item-1>"
294 if multiplicity == "append":
295 header = f"{header}={sample_value}..."
296 else:
297 header = f"{header} {sample_value}..."
298 else:
299 header = f"{header}=<{trait.__class__.__name__}>"
300 # header = "--%s.%s=<%s>" % (cls.__name__, trait.name, trait.__class__.__name__)
301 lines.append(header)
303 if helptext is None:
304 helptext = trait.help
305 if helptext != "":
306 helptext = "\n\n".join(wrap_paragraphs(helptext, 76))
307 lines.append(indent(helptext))
309 if "Enum" in trait.__class__.__name__:
310 # include Enum choices
311 lines.append(indent(f"Choices: {trait.info()}"))
313 if inst is not None:
314 lines.append(indent(f"Current: {getattr(inst, trait.name or '')!r}"))
315 else:
316 try:
317 dvr = trait.default_value_repr()
318 except Exception:
319 dvr = None # ignore defaults we can't construct
320 if dvr is not None:
321 if len(dvr) > 64:
322 dvr = dvr[:61] + "..."
323 lines.append(indent(f"Default: {dvr}"))
325 return "\n".join(lines)
327 @classmethod
328 def class_print_help(cls, inst: HasTraits | None = None) -> None:
329 """Get the help string for a single trait and print it."""
330 print(cls.class_get_help(inst)) # noqa: T201
332 @classmethod
333 def _defining_class(
334 cls, trait: TraitType[t.Any, t.Any], classes: t.Sequence[type[HasTraits]]
335 ) -> type[Configurable]:
336 """Get the class that defines a trait
338 For reducing redundant help output in config files.
339 Returns the current class if:
340 - the trait is defined on this class, or
341 - the class where it is defined would not be in the config file
343 Parameters
344 ----------
345 trait : Trait
346 The trait to look for
347 classes : list
348 The list of other classes to consider for redundancy.
349 Will return `cls` even if it is not defined on `cls`
350 if the defining class is not in `classes`.
351 """
352 defining_cls = cls
353 assert trait.name is not None
354 for parent in cls.mro():
355 if (
356 issubclass(parent, Configurable)
357 and parent in classes
358 and parent.class_own_traits(config=True).get(trait.name, None) is trait
359 ):
360 defining_cls = parent
361 return defining_cls
363 @classmethod
364 def class_config_section(cls, classes: t.Sequence[type[HasTraits]] | None = None) -> str:
365 """Get the config section for this class.
367 Parameters
368 ----------
369 classes : list, optional
370 The list of other classes in the config file.
371 Used to reduce redundant information.
372 """
374 def c(s: str) -> str:
375 """return a commented, wrapped block."""
376 s = "\n\n".join(wrap_paragraphs(s, 78))
378 return "## " + s.replace("\n", "\n# ")
380 # section header
381 breaker = "#" + "-" * 78
382 parent_classes = ", ".join(p.__name__ for p in cls.__bases__ if issubclass(p, Configurable))
384 s = f"# {cls.__name__}({parent_classes}) configuration"
385 lines = [breaker, s, breaker]
386 # get the description trait
387 desc = cls.class_traits().get("description")
388 if desc:
389 desc = desc.default_value
390 if not desc:
391 # no description from trait, use __doc__
392 desc = getattr(cls, "__doc__", "") # type:ignore[arg-type]
393 if desc:
394 lines.append(c(desc)) # type:ignore[arg-type]
395 lines.append("")
397 for name, trait in sorted(cls.class_traits(config=True).items()):
398 default_repr = trait.default_value_repr()
400 if classes:
401 defining_class = cls._defining_class(trait, classes)
402 else:
403 defining_class = cls
404 if defining_class is cls:
405 # cls owns the trait, show full help
406 if trait.help:
407 lines.append(c(trait.help))
408 if "Enum" in type(trait).__name__:
409 # include Enum choices
410 lines.append(f"# Choices: {trait.info()}")
411 lines.append(f"# Default: {default_repr}")
412 else:
413 # Trait appears multiple times and isn't defined here.
414 # Truncate help to first line + "See also Original.trait"
415 if trait.help:
416 lines.append(c(trait.help.split("\n", 1)[0]))
417 lines.append(f"# See also: {defining_class.__name__}.{name}")
419 lines.append(f"# c.{cls.__name__}.{name} = {default_repr}")
420 lines.append("")
421 return "\n".join(lines)
423 @classmethod
424 def class_config_rst_doc(cls) -> str:
425 """Generate rST documentation for this class' config options.
427 Excludes traits defined on parent classes.
428 """
429 lines = []
430 classname = cls.__name__
431 for _, trait in sorted(cls.class_traits(config=True).items()):
432 ttype = trait.__class__.__name__
434 if not trait.name:
435 continue
436 termline = classname + "." + trait.name
438 # Choices or type
439 if "Enum" in ttype:
440 # include Enum choices
441 termline += " : " + trait.info_rst() # type:ignore[attr-defined]
442 else:
443 termline += " : " + ttype
444 lines.append(termline)
446 # Default value
447 try:
448 dvr = trait.default_value_repr()
449 except Exception:
450 dvr = None # ignore defaults we can't construct
451 if dvr is not None:
452 if len(dvr) > 64:
453 dvr = dvr[:61] + "..."
454 # Double up backslashes, so they get to the rendered docs
455 dvr = dvr.replace("\\n", "\\\\n")
456 lines.append(indent(f"Default: ``{dvr}``"))
457 lines.append("")
459 help = trait.help or "No description"
460 lines.append(indent(dedent(help)))
462 # Blank line
463 lines.append("")
465 return "\n".join(lines)
468class LoggingConfigurable(Configurable):
469 """A parent class for Configurables that log.
471 Subclasses have a log trait, and the default behavior
472 is to get the logger from the currently running Application.
473 """
475 log = Any(help="Logger or LoggerAdapter instance", allow_none=False)
477 @validate("log")
478 def _validate_log(self, proposal: Bunch) -> LoggerType:
479 if not isinstance(proposal.value, (logging.Logger, logging.LoggerAdapter)):
480 # warn about unsupported type, but be lenient to allow for duck typing
481 warnings.warn(
482 f"{self.__class__.__name__}.log should be a Logger or LoggerAdapter,"
483 f" got {proposal.value}.",
484 UserWarning,
485 stacklevel=2,
486 )
487 return t.cast(LoggerType, proposal.value)
489 @default("log")
490 def _log_default(self) -> LoggerType:
491 if isinstance(self.parent, LoggingConfigurable):
492 assert self.parent is not None
493 return t.cast(logging.Logger, self.parent.log)
494 from traitlets import log
496 return log.get_logger()
498 def _get_log_handler(self) -> logging.Handler | None:
499 """Return the default Handler
501 Returns None if none can be found
503 Deprecated, this now returns the first log handler which may or may
504 not be the default one.
505 """
506 if not self.log:
507 return None
508 logger: logging.Logger = (
509 self.log if isinstance(self.log, logging.Logger) else self.log.logger
510 )
511 if not getattr(logger, "handlers", None):
512 # no handlers attribute or empty handlers list
513 return None
514 return logger.handlers[0]
517class SingletonConfigurable(LoggingConfigurable):
518 """A configurable that only allows one instance.
520 This class is for classes that should only have one instance of itself
521 or *any* subclass. To create and retrieve such a class use the
522 :meth:`SingletonConfigurable.instance` method.
523 """
525 _instance = None
527 @classmethod
528 def _walk_mro(cls) -> t.Generator[type[SingletonConfigurable], None, None]:
529 """Walk the cls.mro() for parent classes that are also singletons
531 For use in instance()
532 """
534 for subclass in cls.mro():
535 if (
536 issubclass(cls, subclass)
537 and issubclass(subclass, SingletonConfigurable)
538 and subclass != SingletonConfigurable
539 ):
540 yield subclass
542 @classmethod
543 def clear_instance(cls) -> None:
544 """unset _instance for this class and singleton parents."""
545 if not cls.initialized():
546 return
547 for subclass in cls._walk_mro():
548 if isinstance(subclass._instance, cls):
549 # only clear instances that are instances
550 # of the calling class
551 subclass._instance = None # type:ignore[unreachable]
553 @classmethod
554 def instance(cls, *args: t.Any, **kwargs: t.Any) -> Self:
555 """Returns a global instance of this class.
557 This method create a new instance if none have previously been created
558 and returns a previously created instance is one already exists.
560 The arguments and keyword arguments passed to this method are passed
561 on to the :meth:`__init__` method of the class upon instantiation.
563 Examples
564 --------
565 Create a singleton class using instance, and retrieve it::
567 >>> from traitlets.config.configurable import SingletonConfigurable
568 >>> class Foo(SingletonConfigurable): pass
569 >>> foo = Foo.instance()
570 >>> foo == Foo.instance()
571 True
573 Create a subclass that is retrieved using the base class instance::
575 >>> class Bar(SingletonConfigurable): pass
576 >>> class Bam(Bar): pass
577 >>> bam = Bam.instance()
578 >>> bam == Bar.instance()
579 True
580 """
581 # Create and save the instance
582 if cls._instance is None:
583 inst = cls(*args, **kwargs)
584 # Now make sure that the instance will also be returned by
585 # parent classes' _instance attribute.
586 for subclass in cls._walk_mro():
587 subclass._instance = inst
589 if isinstance(cls._instance, cls):
590 return cls._instance
591 else:
592 raise MultipleInstanceError(
593 f"An incompatible sibling of '{cls.__name__}' is already instantiated"
594 f" as singleton: {type(cls._instance).__name__}"
595 )
597 @classmethod
598 def initialized(cls) -> bool:
599 """Has an instance been created?"""
600 return hasattr(cls, "_instance") and cls._instance is not None