Coverage for python / lsst / daf / butler / _storage_class.py: 38%
333 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-17 08:49 +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"""Support for Storage Classes."""
30from __future__ import annotations
32__all__ = ("StorageClass", "StorageClassConfig", "StorageClassFactory")
34import builtins
35import itertools
36import logging
37from collections import ChainMap
38from collections.abc import Callable, Collection, Mapping, Sequence, Set
39from threading import RLock
40from typing import Any
42import pydantic
44from lsst.utils import doImportType
45from lsst.utils.classes import Singleton
46from lsst.utils.introspection import get_full_type_name
48from ._config import Config, ConfigSubset
49from ._config_support import LookupKey
50from ._storage_class_delegate import StorageClassDelegate
52log = logging.getLogger(__name__)
55class StorageClassConfig(ConfigSubset):
56 """Configuration class for defining Storage Classes."""
58 component = "storageClasses"
59 defaultConfigFile = "storageClasses.yaml"
62class _StorageClassModel(pydantic.BaseModel):
63 """Model class used to validate storage class configuration."""
65 pytype: str | None = None
66 inheritsFrom: str | None = None
67 components: dict[str, str] = pydantic.Field(default_factory=dict)
68 derivedComponents: dict[str, str] = pydantic.Field(default_factory=dict)
69 parameters: list[str] = pydantic.Field(default_factory=list)
70 delegate: str | None = None
71 converters: dict[str, str] = pydantic.Field(default_factory=dict)
74class StorageClass:
75 """Class describing how a label maps to a particular Python type.
77 Parameters
78 ----------
79 name : `str`
80 Name to use for this class.
81 pytype : `type` or `str`
82 Python type (or name of type) to associate with the `StorageClass`.
83 components : `dict`, optional
84 `dict` mapping name of a component to another `StorageClass`.
85 derivedComponents : `dict`, optional
86 `dict` mapping name of a derived component to another `StorageClass`.
87 parameters : `~collections.abc.Sequence` or `~collections.abc.Set`
88 Parameters understood by this `StorageClass` that can control
89 reading of data from datastores.
90 delegate : `str`, optional
91 Fully qualified name of class supporting assembly and disassembly
92 of a `pytype` instance.
93 converters : `dict` [`str`, `str`], optional
94 Mapping of python type to function that can be called to convert
95 that python type to the valid type of this storage class.
96 """
98 def __init__(
99 self,
100 name: str = "",
101 pytype: type | str | None = None,
102 components: dict[str, StorageClass] | None = None,
103 derivedComponents: dict[str, StorageClass] | None = None,
104 parameters: Sequence[str] | Set[str] | None = None,
105 delegate: str | None = None,
106 converters: dict[str, str] | None = None,
107 ):
108 # Merge converters with class defaults.
109 self._converters = {}
110 if converters:
111 self._converters.update(converters)
113 # Version of converters where the python types have been
114 # Do not try to import anything until needed.
115 self._converters_by_type: dict[type, Callable[[Any], Any]] | None = None
117 self.name = name
119 if pytype is None:
120 pytype = object
122 self._pytype: type | None
123 if not isinstance(pytype, str):
124 # Already have a type so store it and get the name
125 self._pytypeName = get_full_type_name(pytype)
126 self._pytype = pytype
127 else:
128 # Store the type name and defer loading of type
129 self._pytypeName = pytype
130 self._pytype = None
132 if components is not None:
133 if len(components) == 1: 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true
134 raise ValueError(
135 f"Composite storage class {name} is not allowed to have"
136 f" only one component '{next(iter(components))}'."
137 " Did you mean it to be a derived component?"
138 )
139 self._components = components
140 else:
141 self._components = {}
142 self._derivedComponents = derivedComponents if derivedComponents is not None else {}
143 self._parameters = frozenset(parameters) if parameters is not None else frozenset()
144 # if the delegate is not None also set it and clear the default
145 # delegate
146 self._delegate: type | None
147 self._delegateClassName: str | None
148 if delegate is not None:
149 self._delegateClassName = delegate
150 self._delegate = None
151 elif components is not None:
152 # We set a default delegate for composites so that a class is
153 # guaranteed to support something if it is a composite.
154 log.debug("Setting default delegate for %s", self.name)
155 self._delegate = StorageClassDelegate
156 self._delegateClassName = get_full_type_name(self._delegate)
157 else:
158 self._delegate = None
159 self._delegateClassName = None
161 @property
162 def components(self) -> Mapping[str, StorageClass]:
163 """Return the components associated with this `StorageClass`."""
164 return self._components
166 @property
167 def derivedComponents(self) -> Mapping[str, StorageClass]:
168 """Return derived components associated with `StorageClass`."""
169 return self._derivedComponents
171 @property
172 def converters(self) -> Mapping[str, str]:
173 """Return the type converters supported by this `StorageClass`."""
174 return self._converters
176 def _get_converters_by_type(self) -> Mapping[type, Callable[[Any], Any]]:
177 """Return the type converters as python types."""
178 if self._converters_by_type is None:
179 self._converters_by_type = {}
181 # Loop over list because the dict can be edited in loop.
182 for candidate_type_str, converter_str in list(self.converters.items()):
183 if hasattr(builtins, candidate_type_str):
184 candidate_type = getattr(builtins, candidate_type_str)
185 else:
186 try:
187 candidate_type = doImportType(candidate_type_str)
188 except ImportError as e:
189 log.warning(
190 "Unable to import type %s associated with storage class %s (%s)",
191 candidate_type_str,
192 self.name,
193 e,
194 )
195 del self._converters[candidate_type_str]
196 continue
198 if hasattr(builtins, converter_str):
199 converter = getattr(builtins, converter_str)
200 else:
201 try:
202 converter = doImportType(converter_str)
203 except ImportError as e:
204 log.warning(
205 "Unable to import conversion function %s associated with storage class %s "
206 "required to convert type %s (%s)",
207 converter_str,
208 self.name,
209 candidate_type_str,
210 e,
211 )
212 del self._converters[candidate_type_str]
213 continue
214 if not callable(converter):
215 # doImportType is annotated to return a Type but in actual
216 # fact it can return Any except ModuleType because package
217 # variables can be accessed. This make mypy believe it
218 # is impossible for the return value to not be a callable
219 # so we must ignore the warning.
220 log.warning( # type: ignore
221 "Conversion function %s associated with storage class "
222 "%s to convert type %s is not a callable.",
223 converter_str,
224 self.name,
225 candidate_type_str,
226 )
227 del self._converters[candidate_type_str]
228 continue
229 self._converters_by_type[candidate_type] = converter
230 return self._converters_by_type
232 @property
233 def parameters(self) -> set[str]:
234 """Return `set` of names of supported parameters."""
235 return set(self._parameters)
237 @property
238 def pytype(self) -> type:
239 """Return Python type associated with this `StorageClass`."""
240 if self._pytype is not None:
241 return self._pytype
243 if hasattr(builtins, self._pytypeName):
244 pytype = getattr(builtins, self._pytypeName)
245 else:
246 pytype = doImportType(self._pytypeName)
247 self._pytype = pytype
248 return self._pytype
250 @property
251 def delegateClass(self) -> type | None:
252 """Class to use to delegate type-specific actions."""
253 if self._delegate is not None:
254 return self._delegate
255 if self._delegateClassName is None:
256 return None
257 delegate_class = doImportType(self._delegateClassName)
258 self._delegate = delegate_class
259 return self._delegate
261 def allComponents(self) -> Mapping[str, StorageClass]:
262 """Return all defined components.
264 This mapping includes all the derived and read/write components
265 for the corresponding storage class.
267 Returns
268 -------
269 comp : `dict` of [`str`, `StorageClass`]
270 The component name to storage class mapping.
271 """
272 return ChainMap(self._components, self._derivedComponents)
274 def delegate(self) -> StorageClassDelegate:
275 """Return an instance of a storage class delegate.
277 Returns
278 -------
279 delegate : `StorageClassDelegate`
280 Instance of the delegate associated with this `StorageClass`.
281 The delegate is constructed with this `StorageClass`.
283 Raises
284 ------
285 TypeError
286 This StorageClass has no associated delegate.
287 """
288 cls = self.delegateClass
289 if cls is None:
290 raise TypeError(f"No delegate class is associated with StorageClass {self.name}")
291 return cls(storageClass=self)
293 def isComposite(self) -> bool:
294 """Return Boolean indicating whether this is a composite or not.
296 Returns
297 -------
298 isComposite : `bool`
299 `True` if this `StorageClass` is a composite, `False`
300 otherwise.
301 """
302 if self.components:
303 return True
304 return False
306 def _lookupNames(self) -> tuple[LookupKey, ...]:
307 """Keys to use when looking up this DatasetRef in a configuration.
309 The names are returned in order of priority.
311 Returns
312 -------
313 names : `tuple` of `LookupKey`
314 Tuple of a `LookupKey` using the `StorageClass` name.
315 """
316 return (LookupKey(name=self.name),)
318 def knownParameters(self) -> set[str]:
319 """Return set of all parameters known to this `StorageClass`.
321 The set includes parameters understood by components of a composite.
323 Returns
324 -------
325 known : `set`
326 All parameter keys of this `StorageClass` and the component
327 storage classes.
328 """
329 known = set(self._parameters)
330 for sc in self.components.values():
331 known.update(sc.knownParameters())
332 return known
334 def validateParameters(self, parameters: Collection | None = None) -> None:
335 """Check that the parameters are known to this `StorageClass`.
337 Does not check the values.
339 Parameters
340 ----------
341 parameters : `~collections.abc.Collection`, optional
342 Collection containing the parameters. Can be `dict`-like or
343 `set`-like. The parameter values are not checked.
344 If no parameters are supplied, always returns without error.
346 Raises
347 ------
348 KeyError
349 Some parameters are not understood by this `StorageClass`.
350 """
351 # No parameters is always okay
352 if not parameters:
353 return
355 # Extract the important information into a set. Works for dict and
356 # list.
357 external = set(parameters)
359 diff = external - self.knownParameters()
360 if diff:
361 s = "s" if len(diff) > 1 else ""
362 unknown = "', '".join(diff)
363 raise KeyError(f"Parameter{s} '{unknown}' not understood by StorageClass {self.name}")
365 def filterParameters(
366 self, parameters: Mapping[str, Any] | None, subset: Collection | None = None
367 ) -> Mapping[str, Any]:
368 """Filter out parameters that are not known to this `StorageClass`.
370 Parameters
371 ----------
372 parameters : `~collections.abc.Mapping`, optional
373 Candidate parameters. Can be `None` if no parameters have
374 been provided.
375 subset : `~collections.abc.Collection`, optional
376 Subset of supported parameters that the caller is interested
377 in using. The subset must be known to the `StorageClass`
378 if specified. If `None` the supplied parameters will all
379 be checked, else only the keys in this set will be checked.
381 Returns
382 -------
383 filtered : `~collections.abc.Mapping`
384 Valid parameters. Empty `dict` if none are suitable.
386 Raises
387 ------
388 ValueError
389 Raised if the provided subset is not a subset of the supported
390 parameters or if it is an empty set.
391 """
392 if not parameters:
393 return {}
395 known = self.knownParameters()
397 if subset is not None:
398 if not subset:
399 raise ValueError("Specified a parameter subset but it was empty")
400 subset = set(subset)
401 if not subset.issubset(known):
402 raise ValueError(f"Requested subset ({subset}) is not a subset of known parameters ({known})")
403 wanted = subset
404 else:
405 wanted = known
407 return {k: parameters[k] for k in wanted if k in parameters}
409 def validateInstance(self, instance: Any) -> bool:
410 """Check that the supplied Python object has the expected Python type.
412 Parameters
413 ----------
414 instance : `object`
415 Object to check.
417 Returns
418 -------
419 isOk : `bool`
420 True if the supplied instance object can be handled by this
421 `StorageClass`, False otherwise.
422 """
423 return isinstance(instance, self.pytype)
425 def is_type(self, other: type, compare_types: bool = False) -> bool:
426 """Return Boolean indicating whether the supplied type matches
427 the type in this `StorageClass`.
429 Parameters
430 ----------
431 other : `type`
432 The type to be checked.
433 compare_types : `bool`, optional
434 If `True` the python type will be used in the comparison
435 if the type names do not match. This may trigger an import
436 of code and so can be slower.
438 Returns
439 -------
440 match : `bool`
441 `True` if the types are equal.
443 Notes
444 -----
445 If this `StorageClass` has not yet imported the Python type the
446 check is done against the full type name, this prevents an attempt
447 to import the type when it will likely not match.
448 """
449 if self._pytype:
450 return self._pytype is other
452 other_name = get_full_type_name(other)
453 if self._pytypeName == other_name:
454 return True
456 if compare_types:
457 # Must protect against the import failing.
458 try:
459 return self.pytype is other
460 except Exception:
461 pass
463 return False
465 def can_convert(self, other: StorageClass) -> bool:
466 """Return `True` if this storage class can convert python types
467 in the other storage class.
469 Parameters
470 ----------
471 other : `StorageClass`
472 The storage class to check.
474 Returns
475 -------
476 can : `bool`
477 `True` if this storage class has a registered converter for
478 the python type associated with the other storage class. That
479 converter will convert the other python type to the one associated
480 with this storage class.
481 """
482 if other.name == self.name:
483 # Identical storage classes are compatible.
484 return True
486 # It may be that the storage class being compared is not
487 # available because the python type can't be imported. In that
488 # case conversion must be impossible.
489 try:
490 other_pytype = other.pytype
491 except Exception:
492 return False
494 # Or even this storage class itself can not have the type imported.
495 try:
496 self_pytype = self.pytype
497 except Exception:
498 return False
500 if issubclass(other_pytype, self_pytype):
501 # Storage classes have different names but the same python type.
502 return True
504 for candidate_type in self._get_converters_by_type():
505 if issubclass(other_pytype, candidate_type):
506 return True
507 return False
509 def coerce_type(self, incorrect: Any) -> Any:
510 """Coerce the supplied incorrect instance to the python type
511 associated with this `StorageClass`.
513 Parameters
514 ----------
515 incorrect : `object`
516 An object that might be the incorrect type.
518 Returns
519 -------
520 correct : `object`
521 An object that matches the python type of this `StorageClass`.
522 Can be the same object as given. If `None`, `None` will be
523 returned.
525 Raises
526 ------
527 TypeError
528 Raised if no conversion can be found.
529 """
530 if incorrect is None:
531 return None
533 # Possible this is the correct type already.
534 if self.validateInstance(incorrect):
535 return incorrect
537 # Check each registered converter.
538 for candidate_type, converter in self._get_converters_by_type().items():
539 if isinstance(incorrect, candidate_type):
540 try:
541 return converter(incorrect)
542 except Exception:
543 log.error(
544 "Converter %s failed to convert type %s",
545 get_full_type_name(converter),
546 get_full_type_name(incorrect),
547 )
548 raise
549 raise TypeError(
550 "Type does not match and no valid converter found to convert"
551 f" '{get_full_type_name(incorrect)}' to '{get_full_type_name(self.pytype)}'"
552 )
554 def __eq__(self, other: Any) -> bool:
555 """Equality checks name, pytype name, delegate name, and components."""
556 if not isinstance(other, StorageClass):
557 return NotImplemented
559 if self.name != other.name:
560 return False
562 # We must compare pytype and delegate by name since we do not want
563 # to trigger an import of external module code here
564 if self._delegateClassName != other._delegateClassName:
565 return False
566 if self._pytypeName != other._pytypeName:
567 return False
569 # Ensure we have the same component keys in each
570 if set(self.components.keys()) != set(other.components.keys()):
571 return False
573 # Same parameters
574 if self.parameters != other.parameters:
575 return False
577 # Ensure that all the components have the same type
578 return all(self.components[k] == other.components[k] for k in self.components)
580 def __hash__(self) -> int:
581 return hash(self.name)
583 def __repr__(self) -> str:
584 optionals: dict[str, Any] = {}
585 if self._pytypeName != "object":
586 optionals["pytype"] = self._pytypeName
587 if self._delegateClassName is not None:
588 optionals["delegate"] = self._delegateClassName
589 if self._parameters:
590 optionals["parameters"] = self._parameters
591 if self.components:
592 optionals["components"] = self.components
593 if self.converters:
594 optionals["converters"] = self.converters
596 # order is preserved in the dict
597 options = ", ".join(f"{k}={v!r}" for k, v in optionals.items())
599 # Start with mandatory fields
600 r = f"{self.__class__.__name__}({self.name!r}"
601 if options:
602 r = r + ", " + options
603 r = r + ")"
604 return r
606 def __str__(self) -> str:
607 return self.name
610class StorageClassFactory(metaclass=Singleton):
611 """Factory for `StorageClass` instances.
613 This class is a singleton, with each instance sharing the pool of
614 StorageClasses. Since code can not know whether it is the first
615 time the instance has been created, the constructor takes no arguments.
616 To populate the factory with storage classes, a call to
617 `~StorageClassFactory.addFromConfig()` should be made.
619 Parameters
620 ----------
621 config : `StorageClassConfig` or `str`, optional
622 Load configuration. In a ButlerConfig` the relevant configuration
623 is located in the ``storageClasses`` section.
624 """
626 def __init__(self, config: StorageClassConfig | str | None = None):
627 self._storageClasses: dict[str, StorageClass] = {}
628 self._lock = RLock()
630 # Always seed with the default config
631 self.addFromConfig(StorageClassConfig())
633 if config is not None: 633 ↛ 634line 633 didn't jump to line 634 because the condition on line 633 was never true
634 self.addFromConfig(config)
636 def __str__(self) -> str:
637 """Return summary of factory.
639 Returns
640 -------
641 summary : `str`
642 Summary of the factory status.
643 """
644 with self._lock:
645 sep = "\n"
646 return f"""Number of registered StorageClasses: {len(self._storageClasses)}
648StorageClasses
649--------------
650{sep.join(f"{self._storageClasses[s]!r}" for s in sorted(self._storageClasses))}
651"""
653 def __contains__(self, storageClassOrName: object) -> bool:
654 with self._lock:
655 if isinstance(storageClassOrName, str):
656 return storageClassOrName in self._storageClasses
657 elif isinstance(storageClassOrName, StorageClass):
658 return storageClassOrName.name in self._storageClasses
659 return False
661 def addFromConfig(self, config: StorageClassConfig | Config | str) -> None:
662 """Add more `StorageClass` definitions from a config file.
664 Parameters
665 ----------
666 config : `StorageClassConfig`, `Config` or `str`
667 Storage class configuration. Can contain a ``storageClasses``
668 key if part of a global configuration.
669 """
670 sconfig = StorageClassConfig(config)
672 # Since we can not assume that we will get definitions of
673 # components or parents before their classes are defined
674 # we have a helper function that we can call recursively
675 # to extract definitions from the configuration.
676 def processStorageClass(name: str, _sconfig: StorageClassConfig, msg: str = "") -> StorageClass:
677 # This might have already been processed through recursion, or
678 # already present in the factory.
679 if name not in _sconfig:
680 return self.getStorageClass(name)
681 try:
682 model = _StorageClassModel.model_validate(_sconfig.pop(name))
683 except Exception as err:
684 err.add_note(msg)
685 raise
686 components: dict[str, StorageClass] = {}
687 derivedComponents: dict[str, StorageClass] = {}
688 parameters: set[str] = set()
689 delegate: str | None = None
690 converters: dict[str, str] = {}
691 if model.inheritsFrom is not None:
692 base = processStorageClass(model.inheritsFrom, _sconfig, msg + f"; processing base of {name}")
693 pytype = base._pytypeName
694 components.update(base.components)
695 derivedComponents.update(base.derivedComponents)
696 parameters.update(base.parameters)
697 delegate = base._delegateClassName
698 converters.update(base.converters)
699 if model.pytype is not None: 699 ↛ 701line 699 didn't jump to line 701 because the condition on line 699 was always true
700 pytype = model.pytype
701 for k, v in model.components.items():
702 components[k] = processStorageClass(
703 v, _sconfig, msg + f"; processing component {k} of {name}"
704 )
705 for k, v in model.derivedComponents.items():
706 derivedComponents[k] = processStorageClass(
707 v, _sconfig, msg + f"; processing derivedCmponent {k} of {name}"
708 )
709 parameters.update(model.parameters)
710 if model.delegate is not None:
711 delegate = model.delegate
712 converters.update(model.converters)
713 result = StorageClass(
714 name=name,
715 pytype=pytype,
716 components=components,
717 derivedComponents=derivedComponents,
718 parameters=parameters,
719 delegate=delegate,
720 converters=converters,
721 )
722 self.registerStorageClass(result, msg=msg)
723 return result
725 # In case there is a problem, construct a context message for any
726 # error reporting.
727 files = [str(f) for f in itertools.chain([sconfig.configFile], sconfig.filesRead) if f]
728 context = f"when adding definitions from {', '.join(files)}" if files else ""
729 log.debug("Adding definitions from config %s", ", ".join(files))
731 with self._lock:
732 for name in list(sconfig.keys()):
733 processStorageClass(name, sconfig, context)
735 def getStorageClass(self, storageClassName: str) -> StorageClass:
736 """Get a StorageClass instance associated with the supplied name.
738 Parameters
739 ----------
740 storageClassName : `str`
741 Name of the storage class to retrieve.
743 Returns
744 -------
745 instance : `StorageClass`
746 Instance of the correct `StorageClass`.
748 Raises
749 ------
750 KeyError
751 The requested storage class name is not registered.
752 """
753 with self._lock:
754 return self._storageClasses[storageClassName]
756 def findStorageClass(self, pytype: type, compare_types: bool = False) -> StorageClass:
757 """Find the storage class associated with this python type.
759 Parameters
760 ----------
761 pytype : `type`
762 The Python type to be matched.
763 compare_types : `bool`, optional
764 If `False`, the type will be checked against name of the python
765 type. This comparison is always done first. If `True` and the
766 string comparison failed, each candidate storage class will be
767 forced to have its type imported. This can be significantly slower.
769 Returns
770 -------
771 storageClass : `StorageClass`
772 The matching storage class.
774 Raises
775 ------
776 KeyError
777 Raised if no match could be found.
779 Notes
780 -----
781 It is possible for a python type to be associated with multiple
782 storage classes. This method will currently return the first that
783 matches.
784 """
785 with self._lock:
786 result = self._find_storage_class(pytype, False)
787 if result:
788 return result
790 if compare_types:
791 # The fast comparison failed and we were asked to try the
792 # variant that might involve code imports.
793 result = self._find_storage_class(pytype, True)
794 if result:
795 return result
797 raise KeyError(
798 f"Unable to find a StorageClass associated with type {get_full_type_name(pytype)!r}"
799 )
801 def _find_storage_class(self, pytype: type, compare_types: bool) -> StorageClass | None:
802 """Iterate through all storage classes to find a match.
804 Parameters
805 ----------
806 pytype : `type`
807 The Python type to be matched.
808 compare_types : `bool`, optional
809 Whether to use type name matching or explicit type matching.
810 The latter can be slower.
812 Returns
813 -------
814 storageClass : `StorageClass` or `None`
815 The matching storage class, or `None` if no match was found.
817 Notes
818 -----
819 Helper method for ``findStorageClass``.
820 """
821 with self._lock:
822 for storageClass in self._storageClasses.values():
823 if storageClass.is_type(pytype, compare_types=compare_types):
824 return storageClass
825 return None
827 def registerStorageClass(self, storageClass: StorageClass, msg: str | None = None) -> None:
828 """Store the `StorageClass` in the factory.
830 Will be indexed by `StorageClass.name` and will return instances
831 of the supplied `StorageClass`.
833 Parameters
834 ----------
835 storageClass : `StorageClass`
836 Type of the Python `StorageClass` to register.
837 msg : `str`, optional
838 Additional message string to be included in any error message.
840 Raises
841 ------
842 ValueError
843 If a storage class has already been registered with
844 that storage class name and the previous definition differs.
845 """
846 with self._lock:
847 if storageClass.name in self._storageClasses: 847 ↛ 848line 847 didn't jump to line 848 because the condition on line 847 was never true
848 existing = self.getStorageClass(storageClass.name)
849 if existing != storageClass:
850 errmsg = f" {msg}" if msg else ""
851 raise ValueError(
852 f"New definition for StorageClass {storageClass.name} ({storageClass!r}) "
853 f"differs from current definition ({existing!r}){errmsg}"
854 )
855 if type(existing) is StorageClass and type(storageClass) is not StorageClass:
856 # Replace generic with specialist subclass equivalent.
857 self._storageClasses[storageClass.name] = storageClass
858 else:
859 self._storageClasses[storageClass.name] = storageClass
861 def _unregisterStorageClass(self, storageClassName: str) -> None:
862 """Remove the named StorageClass from the factory.
864 Parameters
865 ----------
866 storageClassName : `str`
867 Name of storage class to remove.
869 Raises
870 ------
871 KeyError
872 The named storage class is not registered.
874 Notes
875 -----
876 This method is intended to simplify testing of StorageClassFactory
877 functionality and it is not expected to be required for normal usage.
878 """
879 with self._lock:
880 del self._storageClasses[storageClassName]
882 def reset(self) -> None:
883 """Remove all storage class entries from factory and reset to
884 initial state.
886 This is useful for test code where a known start state is useful.
887 """
888 with self._lock:
889 self._storageClasses.clear()
890 # Seed with the default config.
891 self.addFromConfig(StorageClassConfig())