Coverage for python/lsst/daf/butler/_config.py: 94%
501 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-16 09:11 +0000
1# This file is part of daf_butler.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (http://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <http://www.gnu.org/licenses/>.
28"""Configuration control."""
30from __future__ import annotations
32__all__ = ("Config", "ConfigSubset")
34import copy
35import io
36import json
37import logging
38import os
39import pprint
40import sys
41from collections import defaultdict
42from collections.abc import Iterable, Iterator, Mapping, MutableMapping, Sequence
43from pathlib import Path
44from typing import IO, TYPE_CHECKING, Any, ClassVar
46import yaml
47from yaml.representer import Representer
49from lsst.resources import ResourcePath, ResourcePathExpression
50from lsst.utils import doImportType
52yaml.add_representer(defaultdict, Representer.represent_dict)
55# Config module logger
56log = logging.getLogger(__name__)
58# PATH-like environment variable to use for defaults.
59CONFIG_PATH = "DAF_BUTLER_CONFIG_PATH"
61if TYPE_CHECKING:
62 yamlLoader = yaml.SafeLoader
63else:
64 try:
65 yamlLoader = yaml.CSafeLoader
66 except AttributeError:
67 # Not all installations have the C library
68 # (but assume for mypy's sake that they're the same)
69 yamlLoader = yaml.SafeLoader
72def _doUpdate(d: Mapping[str, Any], u: Mapping[str, Any]) -> Mapping[str, Any]:
73 if not isinstance(u, Mapping) or not isinstance(d, MutableMapping):
74 raise RuntimeError(f"Only call update with Mapping, not {type(d)}")
75 return _mergeInto(d, u)
78def _mergeInto(d: Any, u: Mapping[str, Any]) -> Any:
79 """Merge ``u`` into ``d`` recursively.
81 Parameters
82 ----------
83 d : `~collections.abc.MutableMapping`
84 Mapping to update in place.
85 u : `~collections.abc.Mapping`
86 Mapping supplying the new values.
88 Returns
89 -------
90 d : `~collections.abc.MutableMapping`
91 The updated mapping.
93 Notes
94 -----
95 Configuration values are almost always plain dictionaries read from YAML,
96 so an exact `dict` check is tried before the abstract base class check.
97 The abstract check is several times more expensive and this runs once per
98 node of every subtree that is merged.
99 """
100 for k, v in u.items():
101 if type(v) is dict or isinstance(v, Mapping):
102 lhs = d.get(k)
103 if type(lhs) is not dict and not isinstance(lhs, MutableMapping):
104 lhs = {}
105 d[k] = _mergeInto(lhs, v)
106 else:
107 d[k] = v
108 return d
111# Types that are common in configurations and are never mappings. Checking
112# membership here is much cheaper than an abstract base class check, and these
113# account for the overwhelming majority of configuration values.
114_NON_MAPPING_TYPES = frozenset({str, int, float, bool, type(None), list, tuple})
117def _copyMapping(u: Mapping[str, Any]) -> dict[str, Any]:
118 """Return a copy of a mapping, recursing into nested mappings.
120 Parameters
121 ----------
122 u : `~collections.abc.Mapping`
123 Mapping to copy.
125 Returns
126 -------
127 copied : `dict`
128 A new `dict`. Nested mappings, including `Config` instances, are
129 copied into plain dictionaries. All other values, including lists, are
130 stored by reference.
132 Notes
133 -----
134 This is equivalent to merging into an empty mapping but skips the
135 look-up and type check of the target that a merge needs, and avoids an
136 abstract base class check for the value types that dominate real
137 configurations.
138 """
139 copied: dict[str, Any] = {}
140 for k, v in u.items():
141 t = type(v)
142 if t is dict:
143 copied[k] = _copyMapping(v)
144 elif t in _NON_MAPPING_TYPES:
145 copied[k] = v
146 elif isinstance(v, Mapping): 146 ↛ 149line 146 didn't jump to line 149 because the condition on line 146 was always true
147 copied[k] = _copyMapping(v)
148 else:
149 copied[k] = v
150 return copied
153def _checkNextItem(k: str | int, d: Any, create: bool, must_be_dict: bool) -> tuple[Any, bool]:
154 """See if k is in d and if it is return the new child."""
155 nextVal = None
156 isThere = False
157 if d is None:
158 # We have gone past the end of the hierarchy
159 pass
160 elif not must_be_dict and isinstance(d, Sequence):
161 # Check for Sequence first because for lists
162 # __contains__ checks whether value is found in list
163 # not whether the index exists in list. When we traverse
164 # the hierarchy we are interested in the index.
165 try:
166 nextVal = d[int(k)]
167 isThere = True
168 except IndexError:
169 pass
170 except ValueError:
171 isThere = k in d
172 elif k in d:
173 nextVal = d[k]
174 isThere = True
175 elif create:
176 d[k] = {}
177 nextVal = d[k]
178 isThere = True
180 return nextVal, isThere
183class Loader(yamlLoader):
184 """YAML Loader that supports file include directives.
186 Uses ``!include`` directive in a YAML file to point to another
187 YAML file to be included. The path in the include directive is relative
188 to the file containing that directive.
190 storageClasses: !include storageClasses.yaml
192 Examples
193 --------
194 >>> with open("document.yaml", "r") as f:
195 data = yaml.load(f, Loader=Loader)
197 Notes
198 -----
199 See https://davidchall.github.io/yaml-includes.html
201 Parameters
202 ----------
203 stream : `str` or `io.IO`
204 The stream to parse.
205 """
207 def __init__(self, stream: str | IO): # types-PyYAML annotates 'stream' with a private type
208 super().__init__(stream)
209 # if this is a string and not a stream we may well lack a name
210 if hasattr(stream, "name"):
211 self._root = ResourcePath(stream.name, forceDirectory=False)
212 else:
213 # No choice but to assume a local filesystem
214 self._root = ResourcePath("no-file.yaml", forceDirectory=False)
215 self.add_constructor("!include", Loader.include)
217 def include(self, node: yaml.Node) -> list[Any] | dict[str, Any]:
218 result: list[Any] | dict[str, Any]
219 if isinstance(node, yaml.ScalarNode):
220 return self.extractFile(self.construct_scalar(node)) # type: ignore[arg-type]
222 elif isinstance(node, yaml.SequenceNode):
223 result = []
224 for filename in self.construct_sequence(node):
225 result.append(self.extractFile(filename))
226 return result
228 elif isinstance(node, yaml.MappingNode): 228 ↛ 237line 228 didn't jump to line 237 because the condition on line 228 was always true
229 result = {}
230 for k, v in self.construct_mapping(node).items():
231 if not isinstance(k, str): 231 ↛ 232line 231 didn't jump to line 232 because the condition on line 231 was never true
232 raise TypeError(f"Expected only strings in YAML mapping; got {k!r} of type {type(k)}.")
233 result[k] = self.extractFile(v)
234 return result
236 else:
237 print("Error:: unrecognised node type in !include statement", file=sys.stderr)
238 raise yaml.constructor.ConstructorError
240 def extractFile(self, filename: str) -> Any:
241 # It is possible for the !include to point to an explicit URI
242 # instead of a relative URI, therefore we first see if it is
243 # scheme-less or not. If it has a scheme we use it directly
244 # if it is scheme-less we use it relative to the file root.
245 requesteduri = ResourcePath(filename, forceAbsolute=False, forceDirectory=False)
247 if requesteduri.scheme:
248 fileuri = requesteduri
249 else:
250 fileuri = self._root.updatedFile(filename)
252 log.debug("Opening YAML file via !include: %s", fileuri)
254 # Read all the data from the resource
255 data = fileuri.read()
257 # Store the bytes into a BytesIO so we can attach a .name
258 stream = io.BytesIO(data)
259 stream.name = fileuri.geturl()
260 return yaml.load(stream, Loader)
263# Type of the key used for accessing items in configuration object. It can be
264# a single string as described below or a sequence of srtings and integer
265# indices. Indices are used to access items in sequences stored in config.
266_ConfigKey = str | Sequence[str | int]
269class Config(MutableMapping):
270 r"""Implements a datatype that is used by `Butler` for configuration.
272 It is essentially a `dict` with key/value pairs, including nested dicts
273 (as values). In fact, it can be initialized with a `dict`.
274 This is explained next:
276 Config extends the `dict` api so that hierarchical values may be accessed
277 with delimited notation or as a tuple. If a string is given the delimiter
278 is picked up from the first character in that string. For example,
279 ``foo.getValue(".a.b.c")``, ``foo["a"]["b"]["c"]``, ``foo["a", "b", "c"]``,
280 ``foo[".a.b.c"]``, and ``foo["/a/b/c"]`` all achieve the same outcome.
281 If the first character is alphanumeric, no delimiter will be used.
282 ``foo["a.b.c"]`` will be a single key ``a.b.c`` as will ``foo[":a.b.c"]``.
283 Unicode characters can be used as the delimiter for distinctiveness if
284 required.
286 If a key in the hierarchy starts with a non-alphanumeric character care
287 should be used to ensure that either the tuple interface is used or
288 a distinct delimiter is always given in string form.
290 Finally, the delimiter can be escaped if it is part of a key and also
291 has to be used as a delimiter. For example, ``foo[r".a.b\.c"]`` results in
292 a two element hierarchy of ``a`` and ``b.c``. For hard-coded strings it is
293 always better to use a different delimiter in these cases.
295 Note that adding a multi-level key implicitly creates any nesting levels
296 that do not exist, but removing multi-level keys does not automatically
297 remove empty nesting levels. As a result:
299 >>> c = Config()
300 >>> c[".a.b"] = 1
301 >>> del c[".a.b"]
302 >>> c["a"]
303 Config({'a': {}})
305 Storage formats supported:
307 - yaml: read and write is supported.
308 - json: read and write is supported but no ``!include`` directive.
310 Parameters
311 ----------
312 other : `lsst.resources.ResourcePath` or `Config` or `dict`
313 Other source of configuration, can be:
315 - (`lsst.resources.ResourcePathExpression`)
316 Treated as a URI to a config file. Must end with ".yaml".
317 - (`Config`) Copies the other Config's values into this one.
318 - (`dict`) Copies the values from the dict into this Config.
320 If `None` is provided an empty `Config` will be created.
321 """
323 _D: str = "→"
324 """Default internal delimiter to use for components in the hierarchy when
325 constructing keys for external use (see `Config.names()`)."""
327 includeKey: ClassVar[str] = "includeConfigs"
328 """Key used to indicate that another config should be included at this
329 part of the hierarchy."""
331 resourcesPackage: str = "lsst.daf.butler"
332 """Package to search for default configuration data. The resources
333 themselves will be within a ``configs`` resource hierarchy."""
335 def __init__(self, other: ResourcePathExpression | Config | Mapping[str, Any] | None = None):
336 self._data: dict[str, Any] = {}
337 self.configFile: ResourcePath | None = None
339 if other is None:
340 return
342 if isinstance(other, Config):
343 # Copying rather than deep copying, because a config entry may
344 # hold an object that cannot be deep copied. Nested mappings are
345 # copied; everything else is stored by reference, which is the
346 # same behaviour a merge into an empty config would give.
347 self._data = _copyMapping(other._data)
348 self.configFile = other.configFile
349 elif isinstance(other, dict | Mapping):
350 # In most cases we have a dict, and it's more efficient
351 # to check for a dict instance before checking the generic mapping.
352 self._data = _copyMapping(other)
353 elif isinstance(other, str | ResourcePath | Path):
354 # if other is a string, assume it is a file path/URI
355 self.__initFromUri(other)
356 self._processExplicitIncludes()
357 else:
358 # if the config specified by other could not be recognized raise
359 # a runtime error.
360 raise RuntimeError(f"A Config could not be loaded from other: {other}")
362 def ppprint(self) -> str:
363 """Return config as formatted readable string.
365 Examples
366 --------
367 use: ``pdb> print(myConfigObject.ppprint())``
369 Returns
370 -------
371 s : `str`
372 A prettyprint formatted string representing the config.
373 """
374 return pprint.pformat(self._data, indent=2, width=1)
376 def __repr__(self) -> str:
377 return f"{type(self).__name__}({self._data!r})"
379 def __str__(self) -> str:
380 return self.ppprint()
382 def __len__(self) -> int:
383 return len(self._data)
385 def __iter__(self) -> Iterator[str]:
386 return iter(self._data)
388 def copy(self) -> Config:
389 return type(self)(self)
391 @classmethod
392 def fromString(cls, string: str, format: str = "yaml") -> Config:
393 """Create a new Config instance from a serialized string.
395 Parameters
396 ----------
397 string : `str`
398 String containing content in specified format.
399 format : `str`, optional
400 Format of the supplied string. Can be ``json`` or ``yaml``.
402 Returns
403 -------
404 c : `Config`
405 Newly-constructed Config.
406 """
407 if format == "yaml":
408 new_config = cls().__initFromYaml(string)
409 elif format == "json":
410 new_config = cls().__initFromJson(string)
411 else:
412 raise ValueError(f"Unexpected format of string: {format}")
413 new_config._processExplicitIncludes()
414 return new_config
416 @classmethod
417 def fromYaml(cls, string: str) -> Config:
418 """Create a new Config instance from a YAML string.
420 Parameters
421 ----------
422 string : `str`
423 String containing content in YAML format.
425 Returns
426 -------
427 c : `Config`
428 Newly-constructed Config.
429 """
430 return cls.fromString(string, format="yaml")
432 def __initFromUri(self, path: ResourcePathExpression) -> None:
433 """Load a file from a path or an URI.
435 Parameters
436 ----------
437 path : `lsst.resources.ResourcePathExpression`
438 Path or a URI to a persisted config file.
439 """
440 uri = ResourcePath(path, forceDirectory=False)
441 ext = uri.getExtension()
442 if ext == ".yaml":
443 log.debug("Opening YAML config file: %s", uri.geturl())
444 content = uri.read()
445 # Use a stream so we can name it
446 stream = io.BytesIO(content)
447 stream.name = uri.geturl()
448 self.__initFromYaml(stream)
449 elif ext == ".json":
450 log.debug("Opening JSON config file: %s", uri.geturl())
451 content = uri.read()
452 self.__initFromJson(content)
453 else:
454 # This URI does not have a valid extension. It might be because
455 # we ended up with a directory and not a file. Before we complain
456 # about an extension, do an existence check. No need to do
457 # the (possibly expensive) existence check in the default code
458 # path above because we will find out soon enough that the file
459 # is not there.
460 if not uri.exists():
461 raise FileNotFoundError(f"Config location {uri} does not exist.")
462 raise RuntimeError(f"The Config URI does not have a supported extension: {uri}")
463 self.configFile = uri
465 def __initFromYaml(self, stream: IO | str | bytes) -> Config:
466 """Load a YAML config from any readable stream that contains one.
468 Parameters
469 ----------
470 stream : `IO` or `str`
471 Stream to pass to the YAML loader. Accepts anything that
472 `yaml.load` accepts. This can include a string as well as an
473 IO stream.
475 Raises
476 ------
477 yaml.YAMLError
478 If there is an error loading the file.
479 """
480 content = yaml.load(stream, Loader=Loader)
481 if content is None:
482 content = {}
483 self._data = content
484 return self
486 def __initFromJson(self, stream: IO | str | bytes) -> Config:
487 """Load a JSON config from any readable stream that contains one.
489 Parameters
490 ----------
491 stream : `IO` or `str`
492 Stream to pass to the JSON loader. This can include a string as
493 well as an IO stream.
495 Raises
496 ------
497 TypeError:
498 Raised if there is an error loading the content.
499 """
500 if isinstance(stream, bytes | str): 500 ↛ 503line 500 didn't jump to line 503 because the condition on line 500 was always true
501 content = json.loads(stream)
502 else:
503 content = json.load(stream)
504 if content is None: 504 ↛ 505line 504 didn't jump to line 505 because the condition on line 504 was never true
505 content = {}
506 self._data = content
507 return self
509 def _processExplicitIncludes(self) -> None:
510 """Scan through the configuration searching for the special includes.
512 Looks for ``includeConfigs`` directive and processes the includes.
513 """
514 # Search paths for config files
515 searchPaths = [ResourcePath(os.path.curdir, forceDirectory=True)]
516 if self.configFile is not None:
517 if isinstance(self.configFile, ResourcePath): 517 ↛ 520line 517 didn't jump to line 520 because the condition on line 517 was always true
518 configDir = self.configFile.dirname()
519 else:
520 raise RuntimeError(f"Unexpected type for config file: {self.configFile}")
521 searchPaths.append(configDir)
523 # Ensure we know what delimiter to use
524 names = self.nameTuples()
525 for path in names:
526 if path[-1] == self.includeKey:
527 log.debug("Processing file include directive at %s", self._D + self._D.join(path))
528 basePath = path[:-1]
530 # Extract the includes and then delete them from the config
531 includes = self[path]
532 del self[path]
534 # Be consistent and convert to a list
535 if not isinstance(includes, list):
536 includes = [includes]
538 # Read each file assuming it is a reference to a file
539 # The file can be relative to config file or cwd
540 # ConfigSubset search paths are not used
541 subConfigs = []
542 for fileName in includes:
543 # Expand any shell variables -- this could be URI
544 fileName = ResourcePath(
545 os.path.expandvars(fileName), forceAbsolute=False, forceDirectory=False
546 )
547 found = None
548 if fileName.isabs():
549 found = fileName
550 else:
551 for dir in searchPaths: 551 ↛ 557line 551 didn't jump to line 557 because the loop on line 551 didn't complete
552 specific = dir.join(fileName.path)
553 # Remote resource check might be expensive
554 if specific.exists():
555 found = specific
556 break
557 if not found: 557 ↛ 558line 557 didn't jump to line 558 because the condition on line 557 was never true
558 raise RuntimeError(f"Unable to find referenced include file: {fileName}")
560 # Read the referenced Config as a Config
561 subConfigs.append(type(self)(found))
563 # Now we need to merge these sub configs with the current
564 # information that was present in this node in the config
565 # tree with precedence given to the explicit values
566 newConfig = subConfigs.pop(0)
567 for sc in subConfigs:
568 newConfig.update(sc)
570 # Explicit values take precedence
571 if not basePath:
572 # This is an include at the root config
573 newConfig.update(self)
574 # Replace the current config
575 self._data = newConfig._data
576 else:
577 newConfig.update(self[basePath])
578 # And reattach to the base config
579 self[basePath] = newConfig
581 @staticmethod
582 def _splitIntoKeys(key: _ConfigKey) -> list[str | int]:
583 r"""Split the argument for get/set/in into a hierarchical list.
585 Parameters
586 ----------
587 key : `str` or iterable
588 Argument given to get/set/in. If an iterable is provided it will
589 be converted to a list. If the first character of the string
590 is not an alphanumeric character then it will be used as the
591 delimiter for the purposes of splitting the remainder of the
592 string. If the delimiter is also in one of the keys then it
593 can be escaped using ``\``. There is no default delimiter.
595 Returns
596 -------
597 keys : `list`
598 Hierarchical keys as a `list`.
599 """
600 if isinstance(key, str):
601 if not key[0].isalnum():
602 d = key[0]
603 key = key[1:]
604 else:
605 return [
606 key,
607 ]
608 escaped = f"\\{d}"
609 temp = None
610 if escaped in key:
611 # Complain at the attempt to escape the escape
612 doubled = rf"\{escaped}"
613 if doubled in key:
614 raise ValueError(
615 f"Escaping an escaped delimiter ({doubled} in {key}) is not yet supported."
616 )
617 # Replace with a character that won't be in the string
618 temp = "\r"
619 if temp in key or d == temp:
620 raise ValueError(
621 f"Can not use character {temp!r} in hierarchical key or as"
622 " delimiter if escaping the delimiter"
623 )
624 key = key.replace(escaped, temp)
625 hierarchy = key.split(d)
626 if temp:
627 hierarchy = [h.replace(temp, d) for h in hierarchy]
628 # Copy the list to keep mypy quiet.
629 return list(hierarchy)
630 elif isinstance(key, Iterable): 630 ↛ 634line 630 didn't jump to line 634 because the condition on line 630 was always true
631 return list(key)
632 else:
633 # Do not try to guess.
634 raise TypeError(f"Provided key [{key}] neither str nor iterable.")
636 def _getKeyHierarchy(self, name: _ConfigKey) -> list[str | int]:
637 """Retrieve the key hierarchy for accessing the Config.
639 Parameters
640 ----------
641 name : `str` or `tuple`
642 Delimited string or `tuple` of hierarchical keys.
644 Returns
645 -------
646 hierarchy : `list` of `str`
647 Hierarchy to use as a `list`. If the name is available directly
648 as a key in the Config it will be used regardless of the presence
649 of any nominal delimiter.
650 """
651 keys: list[str | int]
652 if name in self._data:
653 keys = [name]
654 else:
655 keys = self._splitIntoKeys(name)
656 return keys
658 def _findInHierarchy(self, keys: Sequence[str | int], create: bool = False) -> tuple[list[Any], bool]:
659 """Look for hierarchy of keys in Config.
661 Parameters
662 ----------
663 keys : `list` or `tuple`
664 Keys to search in hierarchy.
665 create : `bool`, optional
666 If `True`, if a part of the hierarchy does not exist, insert an
667 empty `dict` into the hierarchy.
669 Returns
670 -------
671 hierarchy : `list`
672 List of the value corresponding to each key in the supplied
673 hierarchy. Only keys that exist in the hierarchy will have
674 a value.
675 complete : `bool`
676 `True` if the full hierarchy exists and the final element
677 in ``hierarchy`` is the value of relevant value.
678 """
679 d: Any = self._data
681 # For the first key, d must be a dict so it is a waste
682 # of time to check for a sequence.
683 must_be_dict = True
685 hierarchy = []
686 complete = True
687 for k in keys:
688 d, isThere = _checkNextItem(k, d, create, must_be_dict)
689 if isThere:
690 hierarchy.append(d)
691 else:
692 complete = False
693 break
694 # Second time round it might be a sequence.
695 must_be_dict = False
697 return hierarchy, complete
699 def __getitem__(self, name: _ConfigKey) -> Any:
700 # Override the split for the simple case where there is an exact
701 # match. This allows `Config.items()` to work via a simple
702 # __iter__ implementation that returns top level keys of
703 # self._data.
705 # If the name matches a key in the top-level hierarchy, bypass
706 # all further cleverness.
707 found_directly = False
708 try:
709 if isinstance(name, str):
710 data = self._data[name]
711 found_directly = True
712 except KeyError:
713 pass
715 if not found_directly:
716 keys = self._getKeyHierarchy(name)
718 hierarchy, complete = self._findInHierarchy(keys)
719 if not complete:
720 raise KeyError(f"{name} not found")
721 data = hierarchy[-1]
723 # In most cases we have a dict, and it's more efficient
724 # to check for a dict instance before checking the generic mapping.
725 if isinstance(data, dict | Mapping):
726 data = Config(data)
727 # Ensure that child configs inherit the parent internal delimiter
728 if self._D != Config._D:
729 data._D = self._D
730 return data
732 def __setitem__(self, name: _ConfigKey, value: Any) -> None:
733 keys = self._getKeyHierarchy(name)
734 last = keys.pop()
735 if isinstance(value, Config):
736 value = copy.deepcopy(value._data)
738 hierarchy, complete = self._findInHierarchy(keys, create=True)
739 if hierarchy:
740 data = hierarchy[-1]
741 else:
742 data = self._data
744 try:
745 data[last] = value
746 except TypeError:
747 data[int(last)] = value
749 def __contains__(self, key: Any) -> bool:
750 if not isinstance(key, str | Sequence): 750 ↛ 751line 750 didn't jump to line 751 because the condition on line 750 was never true
751 return False
752 keys = self._getKeyHierarchy(key)
753 hierarchy, complete = self._findInHierarchy(keys)
754 return complete
756 def __delitem__(self, key: str | Sequence[str]) -> None:
757 keys = self._getKeyHierarchy(key)
758 last = keys.pop()
759 hierarchy, complete = self._findInHierarchy(keys)
760 if complete: 760 ↛ 767line 760 didn't jump to line 767 because the condition on line 760 was always true
761 if hierarchy:
762 data = hierarchy[-1]
763 else:
764 data = self._data
765 del data[last]
766 else:
767 raise KeyError(f"{key} not found in Config")
769 def update(self, other: Mapping[str, Any]) -> None: # type: ignore[override]
770 """Update config from other `Config` or `dict`.
772 Like `dict.update`, but will add or modify keys in nested dicts,
773 instead of overwriting the nested dict entirely.
775 Parameters
776 ----------
777 other : `dict` or `Config`
778 Source of configuration.
780 Examples
781 --------
782 >>> c = Config({"a": {"b": 1}})
783 >>> c.update({"a": {"c": 2}})
784 >>> print(c)
785 {'a': {'b': 1, 'c': 2}}
787 >>> foo = {"a": {"b": 1}}
788 >>> foo.update({"a": {"c": 2}})
789 >>> print(foo)
790 {'a': {'c': 2}}
791 """
792 _doUpdate(self._data, other)
794 def merge(self, other: Mapping) -> None:
795 """Merge another Config into this one.
797 Like `Config.update()`, but will add keys & values from other that
798 DO NOT EXIST in self.
800 Keys and values that already exist in self will NOT be overwritten.
802 Parameters
803 ----------
804 other : `dict` or `Config`
805 Source of configuration.
806 """
807 if not isinstance(other, Mapping):
808 raise TypeError(f"Can only merge a Mapping into a Config, not {type(other)}")
810 # Convert the supplied mapping to a Config for consistency
811 # This will do a deepcopy if it is already a Config
812 otherCopy = Config(other)
813 otherCopy.update(self)
814 self._data = otherCopy._data
816 def nameTuples(self, topLevelOnly: bool = False) -> list[tuple[str, ...]]:
817 """Get tuples representing the name hierarchies of all keys.
819 The tuples returned from this method are guaranteed to be usable
820 to access items in the configuration object.
822 Parameters
823 ----------
824 topLevelOnly : `bool`, optional
825 If False, the default, a full hierarchy of names is returned.
826 If True, only the top level are returned.
828 Returns
829 -------
830 names : `list` of `tuple` of `str`
831 List of all names present in the `Config` where each element
832 in the list is a `tuple` of strings representing the hierarchy.
833 """
834 if topLevelOnly:
835 return [(k,) for k in self]
837 def getKeysAsTuples(
838 d: Mapping[str, Any] | Sequence[str], keys: list[tuple[str, ...]], base: tuple[str, ...] | None
839 ) -> None:
840 if isinstance(d, Sequence):
841 theseKeys: Iterable[Any] = range(len(d))
842 else:
843 theseKeys = d.keys()
844 for key in theseKeys:
845 val = d[key]
846 levelKey = base + (key,) if base is not None else (key,)
847 keys.append(levelKey)
848 if isinstance(val, Mapping | Sequence) and not isinstance(val, str):
849 getKeysAsTuples(val, keys, levelKey)
851 keys: list[tuple[str, ...]] = []
852 getKeysAsTuples(self._data, keys, None)
853 return keys
855 def names(self, topLevelOnly: bool = False, delimiter: str | None = None) -> list[str]:
856 """Get a delimited name of all the keys in the hierarchy.
858 The values returned from this method are guaranteed to be usable
859 to access items in the configuration object.
861 Parameters
862 ----------
863 topLevelOnly : `bool`, optional
864 If False, the default, a full hierarchy of names is returned.
865 If True, only the top level are returned.
866 delimiter : `str`, optional
867 Delimiter to use when forming the keys. If the delimiter is
868 present in any of the keys, it will be escaped in the returned
869 names. If `None` given a delimiter will be automatically provided.
870 The delimiter can not be alphanumeric.
872 Returns
873 -------
874 names : `list` of `str`
875 List of all names present in the `Config`.
877 Notes
878 -----
879 This is different than the built-in method `dict.keys`, which will
880 return only the first level keys.
882 Raises
883 ------
884 ValueError
885 The supplied delimiter is alphanumeric.
886 """
887 if topLevelOnly:
888 return list(self.keys())
890 # Get all the tuples of hierarchical keys
891 nameTuples = self.nameTuples()
893 if delimiter is not None and delimiter.isalnum(): 893 ↛ 894line 893 didn't jump to line 894 because the condition on line 893 was never true
894 raise ValueError(f"Supplied delimiter ({delimiter!r}) must not be alphanumeric.")
896 if delimiter is None:
897 # Start with something, and ensure it does not need to be
898 # escaped (it is much easier to understand if not escaped)
899 delimiter = self._D
901 # Form big string for easy check of delimiter clash
902 combined = "".join("".join(str(s) for s in k) for k in nameTuples)
904 # Try a delimiter and keep trying until we get something that
905 # works.
906 ntries = 0
907 while delimiter in combined:
908 log.debug("Delimiter '%s' could not be used. Trying another.", delimiter)
909 ntries += 1
911 if ntries > 100: 911 ↛ 912line 911 didn't jump to line 912 because the condition on line 911 was never true
912 raise ValueError(f"Unable to determine a delimiter for Config {self}")
914 # try another one
915 while True:
916 delimiter = chr(ord(delimiter) + 1)
917 if not delimiter.isalnum(): 917 ↛ 915line 917 didn't jump to line 915 because the condition on line 917 was always true
918 break
920 log.debug("Using delimiter %r", delimiter)
922 # Form the keys, escaping the delimiter if necessary
923 strings = [
924 delimiter + delimiter.join(str(s).replace(delimiter, f"\\{delimiter}") for s in k)
925 for k in nameTuples
926 ]
927 return strings
929 def asArray(self, name: str | Sequence[str]) -> Sequence[Any]:
930 """Get a value as an array.
932 May contain one or more elements.
934 Parameters
935 ----------
936 name : `str`
937 Key to use to retrieve value.
939 Returns
940 -------
941 array : `collections.abc.Sequence`
942 The value corresponding to name, but guaranteed to be returned
943 as a list with at least one element. If the value is a
944 `~collections.abc.Sequence` (and not a `str`) the value itself
945 will be returned, else the value will be the first element.
946 """
947 val = self.get(name)
948 if isinstance(val, str) or not isinstance(val, Sequence):
949 val = [val]
950 return val
952 def __eq__(self, other: Any) -> bool:
953 if isinstance(other, Config):
954 other = other._data
955 return self._data == other
957 def __ne__(self, other: Any) -> bool:
958 if isinstance(other, Config):
959 other = other._data
960 return self._data != other
962 #######
963 # i/o #
965 def dump(self, output: IO | None = None, format: str = "yaml") -> str | None:
966 """Write the config to an output stream.
968 Parameters
969 ----------
970 output : `IO`, optional
971 The stream to use for output. If `None` the serialized content
972 will be returned.
973 format : `str`, optional
974 The format to use for the output. Can be "yaml" or "json".
976 Returns
977 -------
978 serialized : `str` or `None`
979 If a stream was given the stream will be used and the return
980 value will be `None`. If the stream was `None` the
981 serialization will be returned as a string.
982 """
983 if format == "yaml":
984 return yaml.safe_dump(self._data, output, default_flow_style=False)
985 elif format == "json": 985 ↛ 991line 985 didn't jump to line 991 because the condition on line 985 was always true
986 if output is not None: 986 ↛ 987line 986 didn't jump to line 987 because the condition on line 986 was never true
987 json.dump(self._data, output, ensure_ascii=False)
988 return None
989 else:
990 return json.dumps(self._data, ensure_ascii=False)
991 raise ValueError(f"Unsupported format for Config serialization: {format}")
993 def dumpToUri(
994 self,
995 uri: ResourcePathExpression,
996 updateFile: bool = True,
997 defaultFileName: str = "butler.yaml",
998 overwrite: bool = True,
999 ) -> None:
1000 """Write the config to location pointed to by given URI.
1002 Currently supports 's3' and 'file' URI schemes.
1004 Parameters
1005 ----------
1006 uri : `lsst.resources.ResourcePathExpression`
1007 URI of location where the Config will be written.
1008 updateFile : bool, optional
1009 If True and uri does not end on a filename with extension, will
1010 append ``defaultFileName`` to the target uri. True by default.
1011 defaultFileName : bool, optional
1012 The file name that will be appended to target uri if updateFile is
1013 True and uri does not end on a file with an extension.
1014 overwrite : bool, optional
1015 If True the configuration will be written even if it already
1016 exists at that location.
1017 """
1018 # Make local copy of URI or create new one
1019 uri = ResourcePath(uri)
1021 if updateFile and not uri.getExtension():
1022 if uri.isdir(): 1022 ↛ 1025line 1022 didn't jump to line 1025 because the condition on line 1022 was always true
1023 uri = uri.join(defaultFileName, forceDirectory=False)
1024 else:
1025 uri = uri.updatedFile(defaultFileName)
1027 # Try to work out the format from the extension
1028 ext = uri.getExtension()
1029 format = ext[1:].lower()
1031 output = self.dump(format=format)
1032 assert output is not None, "Config.dump guarantees not-None return when output arg is None"
1033 uri.write(output.encode(), overwrite=overwrite)
1034 self.configFile = uri
1036 @staticmethod
1037 def updateParameters(
1038 configType: type[ConfigSubset],
1039 config: Config,
1040 full: Config,
1041 toUpdate: dict[str, Any] | None = None,
1042 toCopy: Sequence[str | Sequence[str]] | None = None,
1043 overwrite: bool = True,
1044 toMerge: Sequence[str | Sequence[str]] | None = None,
1045 ) -> None:
1046 """Update specific config parameters.
1048 Allows for named parameters to be set to new values in bulk, and
1049 for other values to be set by copying from a reference config.
1051 Assumes that the supplied config is compatible with ``configType``
1052 and will attach the updated values to the supplied config by
1053 looking for the related component key. It is assumed that
1054 ``config`` and ``full`` are from the same part of the
1055 configuration hierarchy.
1057 Parameters
1058 ----------
1059 configType : `ConfigSubset`
1060 Config type to use to extract relevant items from ``config``.
1061 config : `Config`
1062 A `Config` to update. Only the subset understood by
1063 the supplied `ConfigSubset` will be modified. Default values
1064 will not be inserted and the content will not be validated
1065 since mandatory keys are allowed to be missing until
1066 populated later by merging.
1067 full : `Config`
1068 A complete config with all defaults expanded that can be
1069 converted to a ``configType``. Read-only and will not be
1070 modified by this method. Values are read from here if
1071 ``toCopy`` is defined.
1073 Repository-specific options that should not be obtained
1074 from defaults when Butler instances are constructed
1075 should be copied from ``full`` to ``config``.
1076 toUpdate : `dict`, optional
1077 A `dict` defining the keys to update and the new value to use.
1078 The keys and values can be any supported by `Config`
1079 assignment.
1080 toCopy : `tuple`, optional
1081 `tuple` of keys whose values should be copied from ``full``
1082 into ``config``.
1083 overwrite : `bool`, optional
1084 If `False`, do not modify a value in ``config`` if the key
1085 already exists. Default is always to overwrite.
1086 toMerge : `tuple`, optional
1087 Keys to merge content from full to config without overwriting
1088 pre-existing values. Only works if the key refers to a hierarchy.
1089 The ``overwrite`` flag is ignored.
1091 Raises
1092 ------
1093 ValueError
1094 Neither ``toUpdate``, ``toCopy`` nor ``toMerge`` were defined.
1095 """
1096 if toUpdate is None and toCopy is None and toMerge is None: 1096 ↛ 1097line 1096 didn't jump to line 1097 because the condition on line 1096 was never true
1097 raise ValueError("At least one of toUpdate, toCopy, or toMerge parameters must be set.")
1099 # If this is a parent configuration then we need to ensure that
1100 # the supplied config has the relevant component key in it.
1101 # If this is a parent configuration we add in the stub entry
1102 # so that the ConfigSubset constructor will do the right thing.
1103 # We check full for this since that is guaranteed to be complete.
1104 if (
1105 configType.component is not None
1106 and configType.component in full
1107 and configType.component not in config
1108 ):
1109 config[configType.component] = {}
1111 # Extract the part of the config we wish to update
1112 localConfig = configType(config, mergeDefaults=False, validate=False)
1114 key: str | Sequence[str]
1115 if toUpdate:
1116 for key, value in toUpdate.items():
1117 if key in localConfig and not overwrite:
1118 log.debug(
1119 "Not overriding key '%s' with value '%s' in config %s",
1120 key,
1121 value,
1122 localConfig.__class__.__name__,
1123 )
1124 else:
1125 localConfig[key] = value
1127 if toCopy or toMerge:
1128 localFullConfig = configType(full, mergeDefaults=False)
1130 if toCopy:
1131 for key in toCopy:
1132 if key in localConfig and not overwrite:
1133 log.debug(
1134 "Not overriding key '%s' from defaults in config %s",
1135 key,
1136 localConfig.__class__.__name__,
1137 )
1138 else:
1139 localConfig[key] = localFullConfig[key]
1140 if toMerge:
1141 for key in toMerge:
1142 if key in localConfig:
1143 # Get the node from the config to do the merge
1144 # but then have to reattach to the config.
1145 subset = localConfig[key]
1146 subset.merge(localFullConfig[key])
1147 localConfig[key] = subset
1148 else:
1149 localConfig[key] = localFullConfig[key]
1151 # Reattach to parent if this is a child config
1152 if configType.component is not None and configType.component in config:
1153 config[configType.component] = localConfig
1154 else:
1155 config.update(localConfig)
1157 def toDict(self) -> dict[str, Any]:
1158 """Convert a `Config` to a standalone hierarchical `dict`.
1160 Returns
1161 -------
1162 d : `dict`
1163 The standalone hierarchical `dict` with any `Config` classes
1164 in the hierarchy converted to `dict`.
1166 Notes
1167 -----
1168 This can be useful when passing a Config to some code that
1169 expects native Python types.
1170 """
1171 output = copy.deepcopy(self._data)
1172 for k, v in output.items():
1173 if isinstance(v, Config): 1173 ↛ 1174line 1173 didn't jump to line 1174 because the condition on line 1173 was never true
1174 v = v.toDict()
1175 output[k] = v
1176 return output
1179class ConfigSubset(Config):
1180 """Config representing a subset of a more general configuration.
1182 Subclasses define their own component and when given a configuration
1183 that includes that component, the resulting configuration only includes
1184 the subset. For example, your config might contain ``dimensions`` if it's
1185 part of a global config and that subset will be stored. If ``dimensions``
1186 can not be found it is assumed that the entire contents of the
1187 configuration should be used.
1189 Default values are read from the environment or supplied search paths
1190 using the default configuration file name specified in the subclass.
1191 This allows a configuration class to be instantiated without any
1192 additional arguments.
1194 Additional validation can be specified to check for keys that are mandatory
1195 in the configuration.
1197 Parameters
1198 ----------
1199 other : `Config` or `~lsst.resources.ResourcePathExpression` or `dict`
1200 Argument specifying the configuration information as understood
1201 by `Config`.
1202 validate : `bool`, optional
1203 If `True` required keys will be checked to ensure configuration
1204 consistency.
1205 mergeDefaults : `bool`, optional
1206 If `True` defaults will be read and the supplied config will
1207 be combined with the defaults, with the supplied values taking
1208 precedence.
1209 searchPaths : `list` or `tuple`, optional
1210 Explicit additional paths to search for defaults. They should
1211 be supplied in priority order. These paths have higher priority
1212 than those read from the environment in
1213 `ConfigSubset.defaultSearchPaths()`. Paths can be `str` referring to
1214 the local file system or URIs, `lsst.resources.ResourcePath`.
1215 """
1217 component: ClassVar[str | None] = None
1218 """Component to use from supplied config. Can be None. If specified the
1219 key is not required. Can be a full dot-separated path to a component.
1220 """
1222 requiredKeys: ClassVar[Sequence[str]] = ()
1223 """Keys that are required to be specified in the configuration.
1224 """
1226 defaultConfigFile: ClassVar[str | None] = None
1227 """Name of the file containing defaults for this config class.
1228 """
1230 def __init__(
1231 self,
1232 other: Config | ResourcePathExpression | Mapping[str, Any] | None = None,
1233 validate: bool = True,
1234 mergeDefaults: bool = True,
1235 searchPaths: Sequence[ResourcePathExpression] | None = None,
1236 ):
1237 # Create a blank object to receive the defaults
1238 # Once we have the defaults we then update with the external values
1239 super().__init__()
1241 # Create a standard Config rather than subset
1242 externalConfig = Config(other)
1244 # Select the part we need from it
1245 # To simplify the use of !include we also check for the existence of
1246 # component.component (since the included files can themselves
1247 # include the component name)
1248 if self.component is not None:
1249 doubled = (self.component, self.component)
1250 # Must check for double depth first
1251 if doubled in externalConfig:
1252 externalConfig = externalConfig[doubled]
1253 elif self.component in externalConfig:
1254 externalConfig._data = externalConfig._data[self.component]
1256 # Default files read to create this configuration
1257 self.filesRead: list[ResourcePath | str] = []
1259 # Assume we are not looking up child configurations
1260 containerKey = None
1262 # Sometimes we do not want to merge with defaults.
1263 if mergeDefaults:
1264 # Supplied search paths have highest priority
1265 fullSearchPath: list[ResourcePath | str] = []
1266 if searchPaths:
1267 fullSearchPath = [ResourcePath(path, forceDirectory=True) for path in searchPaths]
1269 # Read default paths from environment
1270 fullSearchPath.extend(self.defaultSearchPaths())
1272 # There are two places to find defaults for this particular config
1273 # - The "defaultConfigFile" defined in the subclass
1274 # - The class specified in the "cls" element in the config.
1275 # Read cls after merging in case it changes.
1276 if self.defaultConfigFile is not None:
1277 self._updateWithConfigsFromPath(fullSearchPath, self.defaultConfigFile)
1279 # Can have a class specification in the external config (priority)
1280 # or from the defaults.
1281 pytype = None
1282 if "cls" in externalConfig:
1283 pytype = externalConfig["cls"]
1284 elif "cls" in self:
1285 pytype = self["cls"]
1287 if pytype is not None:
1288 try:
1289 cls = doImportType(pytype)
1290 except ImportError as e:
1291 raise RuntimeError(f"Failed to import cls '{pytype}' for config {type(self)}") from e
1292 # The class referenced from the config file is not required
1293 # to specify a default config file.
1294 defaultsFile = getattr(cls, "defaultConfigFile", None)
1295 if defaultsFile is not None: 1295 ↛ 1299line 1295 didn't jump to line 1299 because the condition on line 1295 was always true
1296 self._updateWithConfigsFromPath(fullSearchPath, defaultsFile)
1298 # Get the container key in case we need it and it is specified.
1299 containerKey = getattr(cls, "containerKey", None)
1301 # Now update this object with the external values so that the external
1302 # values always override the defaults
1303 self.update(externalConfig)
1304 if not self.configFile: 1304 ↛ 1310line 1304 didn't jump to line 1310 because the condition on line 1304 was always true
1305 self.configFile = externalConfig.configFile
1307 # If this configuration has child configurations of the same
1308 # config class, we need to expand those defaults as well.
1310 if mergeDefaults and containerKey is not None and containerKey in self:
1311 for idx, subConfig in enumerate(self[containerKey]):
1312 self[containerKey, idx] = type(self)(
1313 other=subConfig, validate=validate, mergeDefaults=mergeDefaults, searchPaths=searchPaths
1314 )
1316 if validate:
1317 self.validate()
1319 @classmethod
1320 def defaultSearchPaths(cls) -> list[ResourcePath | str]:
1321 """Read environment to determine search paths to use.
1323 Global defaults, at lowest priority, are found in the ``config``
1324 directory of the butler source tree. Additional defaults can be
1325 defined using the environment variable ``$DAF_BUTLER_CONFIG_PATH``
1326 which is a PATH-like variable where paths at the front of the list
1327 have priority over those later.
1329 Returns
1330 -------
1331 paths : `list`
1332 Returns a list of paths to search. The returned order is in
1333 priority with the highest priority paths first. The butler config
1334 configuration resources will not be included here but will
1335 always be searched last.
1337 Notes
1338 -----
1339 The environment variable is split on the standard ``:`` path separator.
1340 This currently makes it incompatible with usage of URIs.
1341 """
1342 # We can pick up defaults from multiple search paths
1343 # We fill defaults by using the butler config path and then
1344 # the config path environment variable in reverse order.
1345 defaultsPaths: list[str | ResourcePath] = []
1347 if CONFIG_PATH in os.environ:
1348 externalPaths = os.environ[CONFIG_PATH].split(os.pathsep)
1349 defaultsPaths.extend(externalPaths)
1351 # Add the package defaults as a resource
1352 defaultsPaths.append(ResourcePath(f"resource://{cls.resourcesPackage}/configs", forceDirectory=True))
1353 return defaultsPaths
1355 def _updateWithConfigsFromPath(
1356 self, searchPaths: Sequence[str | ResourcePath], configFile: ResourcePath | str
1357 ) -> None:
1358 """Search the supplied paths, merging the configuration values.
1360 The values read will override values currently stored in the object.
1361 Every file found in the path will be read, such that the earlier
1362 path entries have higher priority.
1364 Parameters
1365 ----------
1366 searchPaths : `list` of `lsst.resources.ResourcePath`, `str`
1367 Paths to search for the supplied configFile. This path
1368 is the priority order, such that files read from the
1369 first path entry will be selected over those read from
1370 a later path. Can contain `str` referring to the local file
1371 system or a URI string.
1372 configFile : `lsst.resources.ResourcePath`
1373 File to locate in path. If absolute path it will be read
1374 directly and the search path will not be used. Can be a URI
1375 to an explicit resource (which will ignore the search path)
1376 which is assumed to exist.
1377 """
1378 uri = ResourcePath(configFile, forceDirectory=False)
1379 if uri.isabs() and uri.exists():
1380 # Assume this resource exists
1381 self._updateWithOtherConfigFile(configFile)
1382 self.filesRead.append(configFile)
1383 else:
1384 # Reverse order so that high priority entries
1385 # update the object last.
1386 for pathDir in reversed(searchPaths):
1387 if isinstance(pathDir, str | ResourcePath): 1387 ↛ 1394line 1387 didn't jump to line 1394 because the condition on line 1387 was always true
1388 pathDir = ResourcePath(pathDir, forceDirectory=True)
1389 file = pathDir.join(configFile)
1390 if file.exists():
1391 self.filesRead.append(file)
1392 self._updateWithOtherConfigFile(file)
1393 else:
1394 raise ValueError(f"Unexpected search path type encountered: {pathDir!r}")
1396 def _updateWithOtherConfigFile(self, file: Config | str | ResourcePath | Mapping[str, Any]) -> None:
1397 """Read in some defaults and update.
1399 Update the configuration by reading the supplied file as a config
1400 of this class, and merging such that these values override the
1401 current values. Contents of the external config are not validated.
1403 Parameters
1404 ----------
1405 file : `Config`, `str`, `lsst.resources.ResourcePath`, or `dict`
1406 Entity that can be converted to a `ConfigSubset`.
1407 """
1408 # Use this class to read the defaults so that subsetting can happen
1409 # correctly.
1410 externalConfig = type(self)(file, validate=False, mergeDefaults=False)
1411 self.update(externalConfig)
1413 def validate(self) -> None:
1414 """Check that mandatory keys are present in this configuration.
1416 Ignored if ``requiredKeys`` is empty.
1417 """
1418 # Validation
1419 missing = [k for k in self.requiredKeys if k not in self._data]
1420 if missing:
1421 raise KeyError(f"Mandatory keys ({missing}) missing from supplied configuration for {type(self)}")