Coverage for python / lsst / pex / config / dictField.py: 30%
177 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-28 08:38 +0000
1# This file is part of pex_config.
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__ = ["DictField"]
32import collections.abc
33import weakref
34from collections.abc import Iterator, Mapping
35from typing import Any, ForwardRef, Generic, TypeVar, cast
37from .callStack import StackFrame, getCallStack, getStackFrame
38from .comparison import compareScalars, getComparisonName
39from .config import (
40 Config,
41 Field,
42 FieldValidationError,
43 UnexpectedProxyUsageError,
44 _autocast,
45 _joinNamePath,
46 _typeStr,
47)
49KeyTypeVar = TypeVar("KeyTypeVar")
50ItemTypeVar = TypeVar("ItemTypeVar")
53class Dict(collections.abc.MutableMapping[KeyTypeVar, ItemTypeVar]):
54 """An internal mapping container.
56 This class emulates a `dict`, but adds validation and provenance.
58 Parameters
59 ----------
60 config : `~lsst.pex.config.Config`
61 Config to proxy.
62 field : `~lsst.pex.config.DictField`
63 Field to use.
64 value : `~typing.Any`
65 Value to store.
66 at : `list` of `~lsst.pex.config.callStack.StackFrame`
67 Stack frame for history recording. Will be calculated if `None`.
68 label : `str`, optional
69 Label to use for history recording.
70 setHistory : `bool`, optional
71 Whether to append to the history record.
72 """
74 def __init__(
75 self,
76 config: Config,
77 field: DictField,
78 value: Mapping[KeyTypeVar, ItemTypeVar],
79 *,
80 at: list[StackFrame] | None,
81 label: str,
82 setHistory: bool = True,
83 ):
84 self._field = field
85 self._config_ = weakref.ref(config)
86 self._dict = {}
87 self._history = self._config._history.setdefault(self._field.name, [])
88 self.__doc__ = field.doc
89 if value is not None:
90 try:
91 for k in value:
92 # do not set history per-item
93 self.__setitem__(k, value[k], at=at, label=label, setHistory=False)
94 except TypeError as e:
95 msg = f"Value {value} is of incorrect type {_typeStr(value)}. Mapping type expected."
96 raise FieldValidationError(self._field, self._config, msg) from e
97 if setHistory:
98 self._history.append((dict(self._dict), at, label))
100 @property
101 def _config(self) -> Config:
102 # Config Fields should never outlive their config class instance
103 # assert that as such here
104 value = self._config_()
105 assert value is not None
106 return value
108 history = property(lambda x: x._history)
109 """History (read-only).
110 """
112 def _copy(self, config: Config) -> Dict:
113 return type(self)(config, self._field, self._dict.copy(), at=None, label="copy", setHistory=False)
115 def __getitem__(self, k: KeyTypeVar) -> ItemTypeVar:
116 return self._dict[k]
118 def __len__(self) -> int:
119 return len(self._dict)
121 def __iter__(self) -> Iterator[KeyTypeVar]:
122 return iter(self._dict)
124 def __contains__(self, k: Any) -> bool:
125 return k in self._dict
127 def __setitem__(
128 self, k: KeyTypeVar, x: ItemTypeVar, at: Any = None, label: str = "setitem", setHistory: bool = True
129 ) -> None:
130 if self._config._frozen:
131 msg = f"Cannot modify a frozen Config. Attempting to set item at key {k!r} to value {x}"
132 raise FieldValidationError(self._field, self._config, msg)
134 # validate keytype
135 k = _autocast(k, self._field.keytype)
136 if type(k) is not self._field.keytype:
137 msg = f"Key {k!r} is of type {_typeStr(k)}, expected type {_typeStr(self._field.keytype)}"
138 raise FieldValidationError(self._field, self._config, msg)
140 # validate itemtype
141 x = _autocast(x, self._field.itemtype)
142 if self._field.itemtype is None:
143 if type(x) not in self._field.supportedTypes and x is not None:
144 msg = f"Value {x} at key {k!r} is of invalid type {_typeStr(x)}"
145 raise FieldValidationError(self._field, self._config, msg)
146 else:
147 if type(x) is not self._field.itemtype and x is not None:
148 msg = (
149 f"Value {x} at key {k!r} is of incorrect type {_typeStr(x)}. "
150 f"Expected type {_typeStr(self._field.itemtype)}"
151 )
152 raise FieldValidationError(self._field, self._config, msg)
154 # validate key using keycheck
155 if self._field.keyCheck is not None and not self._field.keyCheck(k):
156 msg = f"Key {k!r} is not a valid key"
157 raise FieldValidationError(self._field, self._config, msg)
159 # validate item using itemcheck
160 if self._field.itemCheck is not None and not self._field.itemCheck(x):
161 msg = f"Item at key {k!r} is not a valid value: {x}"
162 raise FieldValidationError(self._field, self._config, msg)
164 if at is None:
165 at = getCallStack()
167 self._dict[k] = x
168 if setHistory:
169 self._history.append((dict(self._dict), at, label))
171 def __delitem__(
172 self, k: KeyTypeVar, at: Any = None, label: str = "delitem", setHistory: bool = True
173 ) -> None:
174 if self._config._frozen:
175 raise FieldValidationError(self._field, self._config, "Cannot modify a frozen Config")
177 del self._dict[k]
178 if setHistory:
179 if at is None:
180 at = getCallStack()
181 self._history.append((dict(self._dict), at, label))
183 def __repr__(self):
184 return repr(self._dict)
186 def __str__(self):
187 return str(self._dict)
189 def __setattr__(self, attr, value, at=None, label="assignment"):
190 if hasattr(getattr(self.__class__, attr, None), "__set__"):
191 # This allows properties to work.
192 object.__setattr__(self, attr, value)
193 elif attr in self.__dict__ or attr in ["_field", "_config_", "_history", "_dict", "__doc__"]:
194 # This allows specific private attributes to work.
195 object.__setattr__(self, attr, value)
196 else:
197 # We throw everything else.
198 msg = f"{_typeStr(self._field)} has no attribute {attr}"
199 raise FieldValidationError(self._field, self._config, msg)
201 def __reduce__(self):
202 raise UnexpectedProxyUsageError(
203 f"Proxy container for config field {self._field.name} cannot "
204 "be pickled; it should be converted to a built-in container before "
205 "being assigned to other objects or variables."
206 )
209class DictField(Field[Dict[KeyTypeVar, ItemTypeVar]], Generic[KeyTypeVar, ItemTypeVar]):
210 """A configuration field (`~lsst.pex.config.Field` subclass) that maps keys
211 and values.
213 The types of both items and keys are restricted to these builtin types:
214 `int`, `float`, `complex`, `bool`, and `str`). All keys share the same type
215 and all values share the same type. Keys can have a different type from
216 values.
218 Parameters
219 ----------
220 doc : `str`
221 A documentation string that describes the configuration field.
222 keytype : {`int`, `float`, `complex`, `bool`, `str`}, optional
223 The type of the mapping keys. All keys must have this type. Optional
224 if keytype and itemtype are supplied as typing arguments to the class.
225 itemtype : {`int`, `float`, `complex`, `bool`, `str`}, optional
226 Type of the mapping values. Optional if keytype and itemtype are
227 supplied as typing arguments to the class.
228 default : `dict`, optional
229 The default mapping.
230 optional : `bool`, optional
231 If `True`, the field doesn't need to have a set value.
232 dictCheck : `collections.abc.Callable`
233 A function that validates the dictionary as a whole.
234 keyCheck : `collections.abc.Callable`
235 A function that validates individual mapping keys.
236 itemCheck : `collections.abc.Callable`
237 A function that validates individual mapping values.
238 deprecated : `None` or `str`, optional
239 A description of why this Field is deprecated, including removal date.
240 If not `None`, the string is appended to the docstring for this Field.
242 See Also
243 --------
244 ChoiceField
245 ConfigChoiceField
246 ConfigDictField
247 ConfigField
248 ConfigurableField
249 Field
250 ListField
251 RangeField
252 RegistryField
254 Examples
255 --------
256 This field maps has `str` keys and `int` values:
258 >>> from lsst.pex.config import Config, DictField
259 >>> class MyConfig(Config):
260 ... field = DictField(
261 ... doc="Example string-to-int mapping field.",
262 ... keytype=str,
263 ... itemtype=int,
264 ... default={},
265 ... )
266 >>> config = MyConfig()
267 >>> config.field["myKey"] = 42
268 >>> print(config.field)
269 {'myKey': 42}
270 """
272 DictClass: type[Dict] = Dict
274 @staticmethod
275 def _parseTypingArgs(
276 params: tuple[type, ...] | tuple[str, ...], kwds: Mapping[str, Any]
277 ) -> Mapping[str, Any]:
278 if len(params) != 2: 278 ↛ 279line 278 didn't jump to line 279 because the condition on line 278 was never true
279 raise ValueError("Only tuples of types that are length 2 are supported")
280 resultParams = []
281 for typ in params:
282 if isinstance(typ, str): 282 ↛ 283line 282 didn't jump to line 283 because the condition on line 282 was never true
283 _typ = ForwardRef(typ)
284 # type ignore below because typeshed seems to be wrong. It
285 # indicates there are only 2 args, as it was in python 3.8, but
286 # 3.9+ takes 3 args.
287 result = _typ._evaluate(globals(), locals(), recursive_guard=set()) # type: ignore
288 if result is None:
289 raise ValueError("Could not deduce type from input")
290 typ = cast(type, result)
291 resultParams.append(typ)
292 keyType, itemType = resultParams
293 results = dict(kwds)
294 if (supplied := kwds.get("keytype")) and supplied != keyType: 294 ↛ 295line 294 didn't jump to line 295 because the condition on line 294 was never true
295 raise ValueError("Conflicting definition for keytype")
296 else:
297 results["keytype"] = keyType
298 if (supplied := kwds.get("itemtype")) and supplied != itemType: 298 ↛ 299line 298 didn't jump to line 299 because the condition on line 298 was never true
299 raise ValueError("Conflicting definition for itemtype")
300 else:
301 results["itemtype"] = itemType
302 return results
304 def __init__(
305 self,
306 doc,
307 keytype=None,
308 itemtype=None,
309 default=None,
310 optional=False,
311 dictCheck=None,
312 keyCheck=None,
313 itemCheck=None,
314 deprecated=None,
315 ):
316 source = getStackFrame()
317 self._setup(
318 doc=doc,
319 dtype=Dict,
320 default=default,
321 check=None,
322 optional=optional,
323 source=source,
324 deprecated=deprecated,
325 )
326 if keytype is None: 326 ↛ 327line 326 didn't jump to line 327 because the condition on line 326 was never true
327 raise ValueError(
328 "keytype must either be supplied as an argument or as a type argument to the class"
329 )
330 if keytype not in self.supportedTypes: 330 ↛ 331line 330 didn't jump to line 331 because the condition on line 330 was never true
331 raise ValueError(f"'keytype' {_typeStr(keytype)} is not a supported type")
332 elif itemtype is not None and itemtype not in self.supportedTypes: 332 ↛ 333line 332 didn't jump to line 333 because the condition on line 332 was never true
333 raise ValueError(f"'itemtype' {_typeStr(itemtype)} is not a supported type")
335 check_errors = []
336 for name, check in (("dictCheck", dictCheck), ("keyCheck", keyCheck), ("itemCheck", itemCheck)):
337 if check is not None and not callable(check): 337 ↛ 338line 337 didn't jump to line 338 because the condition on line 337 was never true
338 check_errors.append(name)
339 if check_errors: 339 ↛ 340line 339 didn't jump to line 340 because the condition on line 339 was never true
340 raise ValueError(f"{', '.join(check_errors)} must be callable")
342 self.keytype = keytype
343 self.itemtype = itemtype
344 self.dictCheck = dictCheck
345 self.keyCheck = keyCheck
346 self.itemCheck = itemCheck
348 def validate(self, instance):
349 """Validate the field's value (for internal use only).
351 Parameters
352 ----------
353 instance : `lsst.pex.config.Config`
354 The configuration that contains this field.
356 Raises
357 ------
358 lsst.pex.config.FieldValidationError
359 Raised if validation fails for this field (see *Notes*).
361 Notes
362 -----
363 This method validates values according to the following criteria:
365 - A non-optional field is not `None`.
366 - If a value is not `None`, it must pass the `ConfigField.dictCheck`
367 user callback function.
369 Individual key and item checks by the ``keyCheck`` and ``itemCheck``
370 user callback functions are done immediately when the value is set on a
371 key. Those checks are not repeated by this method.
372 """
373 Field.validate(self, instance)
374 value = self.__get__(instance)
375 if value is not None and self.dictCheck is not None and not self.dictCheck(value):
376 msg = f"{value} is not a valid value"
377 raise FieldValidationError(self, instance, msg)
379 def __set__(
380 self,
381 instance: Config,
382 value: Mapping[KeyTypeVar, ItemTypeVar] | None,
383 at: Any = None,
384 label: str = "assignment",
385 ) -> None:
386 if instance._frozen:
387 msg = f"Cannot modify a frozen Config. Attempting to set field to value {value}"
388 raise FieldValidationError(self, instance, msg)
390 if at is None:
391 at = getCallStack()
392 if value is not None:
393 value = self.DictClass(instance, self, value, at=at, label=label)
394 else:
395 history = instance._history.setdefault(self.name, [])
396 history.append((value, at, label))
398 instance._storage[self.name] = value
400 def toDict(self, instance):
401 """Convert this field's key-value pairs into a regular `dict`.
403 Parameters
404 ----------
405 instance : `lsst.pex.config.Config`
406 The configuration that contains this field.
408 Returns
409 -------
410 result : `dict` or `None`
411 If this field has a value of `None`, then this method returns
412 `None`. Otherwise, this method returns the field's value as a
413 regular Python `dict`.
414 """
415 value = self.__get__(instance)
416 return dict(value) if value is not None else None
418 def _copy_storage(self, old: Config, new: Config) -> Dict[KeyTypeVar, ItemTypeVar] | None:
419 value: Dict[KeyTypeVar, ItemTypeVar] | None = old._storage[self.name]
420 if value is not None:
421 return value._copy(new)
422 else:
423 return None
425 def _compare(self, instance1, instance2, shortcut, rtol, atol, output):
426 """Compare two fields for equality.
428 Used by `lsst.pex.ConfigDictField.compare`.
430 Parameters
431 ----------
432 instance1 : `lsst.pex.config.Config`
433 Left-hand side config instance to compare.
434 instance2 : `lsst.pex.config.Config`
435 Right-hand side config instance to compare.
436 shortcut : `bool`
437 If `True`, this function returns as soon as an inequality if found.
438 rtol : `float`
439 Relative tolerance for floating point comparisons.
440 atol : `float`
441 Absolute tolerance for floating point comparisons.
442 output : `collections.abc.Callable`
443 A callable that takes a string, used (possibly repeatedly) to
444 report inequalities.
446 Returns
447 -------
448 isEqual : bool
449 `True` if the fields are equal, `False` otherwise.
451 Notes
452 -----
453 Floating point comparisons are performed by `numpy.allclose`.
454 """
455 d1 = getattr(instance1, self.name)
456 d2 = getattr(instance2, self.name)
457 name = getComparisonName(
458 _joinNamePath(instance1._name, self.name), _joinNamePath(instance2._name, self.name)
459 )
460 if d1 is None or d2 is None:
461 return compareScalars(name, d1, d2, output=output)
462 if not compareScalars(f"{name} (keys)", set(d1.keys()), set(d2.keys()), output=output):
463 return False
464 equal = True
465 for k, v1 in d1.items():
466 v2 = d2[k]
467 result = compareScalars(
468 f"{name}[{k!r}]", v1, v2, dtype=self.itemtype, rtol=rtol, atol=atol, output=output
469 )
470 if not result and shortcut:
471 return False
472 equal = equal and result
473 return equal