Coverage for python / lsst / daf / butler / mapping_factory.py: 18%
76 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-06 08:30 +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/>.
28from __future__ import annotations
30__all__ = ("MappingFactory",)
32from collections.abc import Iterable
33from typing import Any
35from lsst.utils.introspection import get_class_of
37from ._config import Config
38from ._config_support import LookupKey
41class MappingFactory:
42 """Register the mapping of some key to a python type and retrieve
43 instances.
45 Enables instances of these classes to be retrieved from the factory later.
46 The class can be specified as an object, class or string.
47 If the key is an object it is converted to a string by accessing
48 a ``name`` attribute.
50 Parameters
51 ----------
52 refType : `type`
53 Python reference `type` to use to ensure that items stored in the
54 registry create instance objects of the correct class. Subclasses
55 of this type are allowed. Using `None` disables the check.
56 """
58 def __init__(self, refType: type):
59 self._registry: dict[LookupKey, dict[str, Any]] = {}
60 self.refType = refType
62 def __contains__(self, key: Any) -> bool:
63 """Indicate whether the supplied key is present in the factory.
65 Parameters
66 ----------
67 key : `LookupKey`, `str` or objects with ``name`` attribute
68 Key to use to lookup whether a corresponding element exists
69 in this factory.
71 Returns
72 -------
73 in : `bool`
74 `True` if the supplied key is present in the factory.
75 """
76 key = self._getNameKey(key)
77 return key in self._registry
79 def getLookupKeys(self) -> set[LookupKey]:
80 """Retrieve the look up keys for all the registry entries.
82 Returns
83 -------
84 keys : `set` of `LookupKey`
85 The keys available for matching in the registry.
86 """
87 return set(self._registry)
89 def getClassFromRegistryWithMatch(
90 self, targetClasses: Iterable[Any]
91 ) -> tuple[LookupKey, type, dict[Any, Any]]:
92 """Get the class stored in the registry along with the matching key.
94 Parameters
95 ----------
96 targetClasses : `LookupKey`, `str` or objects with ``name`` attribute
97 Each item is tested in turn until a match is found in the registry.
98 Items with `None` value are skipped.
100 Returns
101 -------
102 matchKey : `LookupKey`
103 The key that resulted in the successful match.
104 cls : `type`
105 Class stored in registry associated with the first
106 matching target class.
107 kwargs: `dict`
108 Keyword arguments to be given to constructor.
110 Raises
111 ------
112 KeyError
113 Raised if none of the supplied target classes match an item in the
114 registry.
115 """
116 attempts: list[Any] = []
117 for t in targetClasses:
118 if t is None:
119 attempts.append(t)
120 else:
121 key = self._getNameKey(t)
122 attempts.append(key)
123 try:
124 entry = self._registry[key]
125 except KeyError:
126 pass
127 else:
128 return key, get_class_of(entry["type"]), entry["kwargs"]
130 # Convert list to a string for error reporting
131 msg = ", ".join(str(k) for k in attempts)
132 plural = "" if len(attempts) == 1 else "s"
133 raise KeyError(f"Unable to find item in registry with key{plural}: {msg}")
135 def getClassFromRegistry(self, targetClasses: Iterable[Any]) -> type:
136 """Get the matching class stored in the registry.
138 Parameters
139 ----------
140 targetClasses : `LookupKey`, `str` or objects with ``name`` attribute
141 Each item is tested in turn until a match is found in the registry.
142 Items with `None` value are skipped.
144 Returns
145 -------
146 cls : `type`
147 Class stored in registry associated with the first
148 matching target class.
150 Raises
151 ------
152 KeyError
153 Raised if none of the supplied target classes match an item in the
154 registry.
155 """
156 _, cls, _ = self.getClassFromRegistryWithMatch(targetClasses)
157 return cls
159 def getFromRegistryWithMatch(
160 self, targetClasses: Iterable[Any], *args: Any, **kwargs: Any
161 ) -> tuple[LookupKey, Any]:
162 """Get a new instance of the registry object along with matching key.
164 Parameters
165 ----------
166 targetClasses : `LookupKey`, `str` or objects with ``name`` attribute
167 Each item is tested in turn until a match is found in the registry.
168 Items with `None` value are skipped.
169 *args : `tuple`
170 Positional arguments to use pass to the object constructor.
171 **kwargs
172 Keyword arguments to pass to object constructor.
174 Returns
175 -------
176 matchKey : `LookupKey`
177 The key that resulted in the successful match.
178 instance : `object`
179 Instance of class stored in registry associated with the first
180 matching target class.
182 Raises
183 ------
184 KeyError
185 Raised if none of the supplied target classes match an item in the
186 registry.
187 """
188 key, cls, registry_kwargs = self.getClassFromRegistryWithMatch(targetClasses)
190 # Supplied keyword args must overwrite registry defaults
191 # We want this overwriting to happen recursively since we expect
192 # some of these keyword arguments to be dicts.
193 # Simplest to use Config for this but have to be careful for top-level
194 # objects that look like Mappings but aren't (eg DataCoordinate) so
195 # only merge keys that appear in both kwargs.
196 merged_kwargs: dict[str, Any] = {}
197 if kwargs and registry_kwargs:
198 kwargs_keys = set(kwargs.keys())
199 registry_keys = set(registry_kwargs.keys())
200 common_keys = kwargs_keys & registry_keys
201 if common_keys:
202 config_kwargs = Config({k: registry_kwargs[k] for k in common_keys})
203 config_kwargs.update({k: kwargs[k] for k in common_keys})
204 merged_kwargs = config_kwargs.toDict()
206 if kwargs_only := kwargs_keys - common_keys:
207 merged_kwargs.update({k: kwargs[k] for k in kwargs_only})
208 if registry_only := registry_keys - common_keys:
209 merged_kwargs.update({k: registry_kwargs[k] for k in registry_only})
210 else:
211 # Distinct in each.
212 merged_kwargs = kwargs
213 merged_kwargs.update(registry_kwargs)
214 elif registry_kwargs:
215 merged_kwargs = registry_kwargs
216 elif kwargs:
217 merged_kwargs = kwargs
219 return key, cls(*args, **merged_kwargs)
221 def getFromRegistry(self, targetClasses: Iterable[Any], *args: Any, **kwargs: Any) -> Any:
222 """Get a new instance of the object stored in the registry.
224 Parameters
225 ----------
226 targetClasses : `LookupKey`, `str` or objects with ``name`` attribute
227 Each item is tested in turn until a match is found in the registry.
228 Items with `None` value are skipped.
229 *args : `tuple`
230 Positional arguments to use pass to the object constructor.
231 **kwargs
232 Keyword arguments to pass to object constructor.
234 Returns
235 -------
236 instance : `object`
237 Instance of class stored in registry associated with the first
238 matching target class.
240 Raises
241 ------
242 KeyError
243 Raised if none of the supplied target classes match an item in the
244 registry.
245 """
246 _, instance = self.getFromRegistryWithMatch(targetClasses, *args, **kwargs)
247 return instance
249 def placeInRegistry(
250 self, registryKey: Any, typeName: str | type, overwrite: bool = False, **kwargs: Any
251 ) -> None:
252 """Register a class name with the associated type.
254 Parameters
255 ----------
256 registryKey : `LookupKey`, `str` or object with ``name`` attribute
257 Item to associate with the provided type.
258 typeName : `str` or Python type
259 Identifies a class to associate with the provided key.
260 overwrite : `bool`, optional
261 If `True`, an existing entry will be overwritten. This option
262 is expected to be used to simplify test suites.
263 Default is `False`.
264 **kwargs
265 Keyword arguments to always pass to object constructor when
266 retrieved.
268 Raises
269 ------
270 KeyError
271 Raised if item is already registered and has different value and
272 ``overwrite`` is `False`.
273 """
274 key = self._getNameKey(registryKey)
275 if key in self._registry and not overwrite:
276 # Compare the class strings since dynamic classes can be the
277 # same thing but be different.
278 if str(self._registry[key]) == str(typeName):
279 return
281 raise KeyError(
282 f"Item with key {key} already registered with different value "
283 f"({self._registry[key]} != {typeName})"
284 )
286 self._registry[key] = {
287 "type": typeName,
288 "kwargs": dict(**kwargs),
289 }
291 @staticmethod
292 def _getNameKey(typeOrName: Any) -> LookupKey:
293 """Extract name of supplied object as entity suitable for key use.
295 Parameters
296 ----------
297 typeOrName : `LookupKey, `str` or object supporting ``name`` attribute.
298 Item from which to extract a name.
300 Returns
301 -------
302 name : `LookupKey`
303 Extracted name as a string or
304 """
305 if isinstance(typeOrName, LookupKey):
306 return typeOrName
308 if isinstance(typeOrName, str):
309 name = typeOrName
310 elif hasattr(typeOrName, "name"):
311 name = typeOrName.name
312 else:
313 raise ValueError("Cannot extract name from type")
315 return LookupKey(name=name)