Coverage for python/lsst/pex/config/dictField.py: 31%
170 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-10-09 03:27 -0700
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 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__(self, config, field, value, at, label, setHistory=True):
75 self._field = field
76 self._config_ = weakref.ref(config)
77 self._dict = {}
78 self._history = self._config._history.setdefault(self._field.name, [])
79 self.__doc__ = field.doc
80 if value is not None:
81 try:
82 for k in value:
83 # do not set history per-item
84 self.__setitem__(k, value[k], at=at, label=label, setHistory=False)
85 except TypeError as e:
86 msg = f"Value {value} is of incorrect type {_typeStr(value)}. Mapping type expected."
87 raise FieldValidationError(self._field, self._config, msg) from e
88 if setHistory:
89 self._history.append((dict(self._dict), at, label))
91 @property
92 def _config(self) -> Config:
93 # Config Fields should never outlive their config class instance
94 # assert that as such here
95 value = self._config_()
96 assert value is not None
97 return value
99 history = property(lambda x: x._history)
100 """History (read-only).
101 """
103 def __getitem__(self, k: KeyTypeVar) -> ItemTypeVar:
104 return self._dict[k]
106 def __len__(self) -> int:
107 return len(self._dict)
109 def __iter__(self) -> Iterator[KeyTypeVar]:
110 return iter(self._dict)
112 def __contains__(self, k: Any) -> bool:
113 return k in self._dict
115 def __setitem__(
116 self, k: KeyTypeVar, x: ItemTypeVar, at: Any = None, label: str = "setitem", setHistory: bool = True
117 ) -> None:
118 if self._config._frozen:
119 msg = f"Cannot modify a frozen Config. Attempting to set item at key {k!r} to value {x}"
120 raise FieldValidationError(self._field, self._config, msg)
122 # validate keytype
123 k = _autocast(k, self._field.keytype)
124 if type(k) is not self._field.keytype:
125 msg = f"Key {k!r} is of type {_typeStr(k)}, expected type {_typeStr(self._field.keytype)}"
126 raise FieldValidationError(self._field, self._config, msg)
128 # validate itemtype
129 x = _autocast(x, self._field.itemtype)
130 if self._field.itemtype is None:
131 if type(x) not in self._field.supportedTypes and x is not None:
132 msg = f"Value {x} at key {k!r} is of invalid type {_typeStr(x)}"
133 raise FieldValidationError(self._field, self._config, msg)
134 else:
135 if type(x) is not self._field.itemtype and x is not None:
136 msg = (
137 f"Value {x} at key {k!r} is of incorrect type {_typeStr(x)}. "
138 f"Expected type {_typeStr(self._field.itemtype)}"
139 )
140 raise FieldValidationError(self._field, self._config, msg)
142 # validate key using keycheck
143 if self._field.keyCheck is not None and not self._field.keyCheck(k):
144 msg = f"Key {k!r} is not a valid key"
145 raise FieldValidationError(self._field, self._config, msg)
147 # validate item using itemcheck
148 if self._field.itemCheck is not None and not self._field.itemCheck(x):
149 msg = f"Item at key {k!r} is not a valid value: {x}"
150 raise FieldValidationError(self._field, self._config, msg)
152 if at is None:
153 at = getCallStack()
155 self._dict[k] = x
156 if setHistory:
157 self._history.append((dict(self._dict), at, label))
159 def __delitem__(
160 self, k: KeyTypeVar, at: Any = None, label: str = "delitem", setHistory: bool = True
161 ) -> None:
162 if self._config._frozen:
163 raise FieldValidationError(self._field, self._config, "Cannot modify a frozen Config")
165 del self._dict[k]
166 if setHistory:
167 if at is None:
168 at = getCallStack()
169 self._history.append((dict(self._dict), at, label))
171 def __repr__(self):
172 return repr(self._dict)
174 def __str__(self):
175 return str(self._dict)
177 def __setattr__(self, attr, value, at=None, label="assignment"):
178 if hasattr(getattr(self.__class__, attr, None), "__set__"):
179 # This allows properties to work.
180 object.__setattr__(self, attr, value)
181 elif attr in self.__dict__ or attr in ["_field", "_config_", "_history", "_dict", "__doc__"]:
182 # This allows specific private attributes to work.
183 object.__setattr__(self, attr, value)
184 else:
185 # We throw everything else.
186 msg = f"{_typeStr(self._field)} has no attribute {attr}"
187 raise FieldValidationError(self._field, self._config, msg)
189 def __reduce__(self):
190 raise UnexpectedProxyUsageError(
191 f"Proxy container for config field {self._field.name} cannot "
192 "be pickled; it should be converted to a built-in container before "
193 "being assigned to other objects or variables."
194 )
197class DictField(Field[Dict[KeyTypeVar, ItemTypeVar]], Generic[KeyTypeVar, ItemTypeVar]):
198 """A configuration field (`~lsst.pex.config.Field` subclass) that maps keys
199 and values.
201 The types of both items and keys are restricted to these builtin types:
202 `int`, `float`, `complex`, `bool`, and `str`). All keys share the same type
203 and all values share the same type. Keys can have a different type from
204 values.
206 Parameters
207 ----------
208 doc : `str`
209 A documentation string that describes the configuration field.
210 keytype : {`int`, `float`, `complex`, `bool`, `str`}, optional
211 The type of the mapping keys. All keys must have this type. Optional
212 if keytype and itemtype are supplied as typing arguments to the class.
213 itemtype : {`int`, `float`, `complex`, `bool`, `str`}, optional
214 Type of the mapping values. Optional if keytype and itemtype are
215 supplied as typing arguments to the class.
216 default : `dict`, optional
217 The default mapping.
218 optional : `bool`, optional
219 If `True`, the field doesn't need to have a set value.
220 dictCheck : callable
221 A function that validates the dictionary as a whole.
222 keyCheck : callable
223 A function that validates individual mapping keys.
224 itemCheck : callable
225 A function that validates individual mapping values.
226 deprecated : None or `str`, optional
227 A description of why this Field is deprecated, including removal date.
228 If not None, the string is appended to the docstring for this Field.
230 See Also
231 --------
232 ChoiceField
233 ConfigChoiceField
234 ConfigDictField
235 ConfigField
236 ConfigurableField
237 Field
238 ListField
239 RangeField
240 RegistryField
242 Examples
243 --------
244 This field maps has `str` keys and `int` values:
246 >>> from lsst.pex.config import Config, DictField
247 >>> class MyConfig(Config):
248 ... field = DictField(
249 ... doc="Example string-to-int mapping field.",
250 ... keytype=str,
251 ... itemtype=int,
252 ... default={},
253 ... )
254 >>> config = MyConfig()
255 >>> config.field["myKey"] = 42
256 >>> print(config.field)
257 {'myKey': 42}
258 """
260 DictClass: type[Dict] = Dict
262 @staticmethod
263 def _parseTypingArgs(
264 params: tuple[type, ...] | tuple[str, ...], kwds: Mapping[str, Any]
265 ) -> Mapping[str, Any]:
266 if len(params) != 2: 266 ↛ 267line 266 didn't jump to line 267 because the condition on line 266 was never true
267 raise ValueError("Only tuples of types that are length 2 are supported")
268 resultParams = []
269 for typ in params:
270 if isinstance(typ, str): 270 ↛ 271line 270 didn't jump to line 271 because the condition on line 270 was never true
271 _typ = ForwardRef(typ)
272 # type ignore below because typeshed seems to be wrong. It
273 # indicates there are only 2 args, as it was in python 3.8, but
274 # 3.9+ takes 3 args.
275 result = _typ._evaluate(globals(), locals(), recursive_guard=set()) # type: ignore
276 if result is None:
277 raise ValueError("Could not deduce type from input")
278 typ = cast(type, result)
279 resultParams.append(typ)
280 keyType, itemType = resultParams
281 results = dict(kwds)
282 if (supplied := kwds.get("keytype")) and supplied != keyType: 282 ↛ 283line 282 didn't jump to line 283 because the condition on line 282 was never true
283 raise ValueError("Conflicting definition for keytype")
284 else:
285 results["keytype"] = keyType
286 if (supplied := kwds.get("itemtype")) and supplied != itemType: 286 ↛ 287line 286 didn't jump to line 287 because the condition on line 286 was never true
287 raise ValueError("Conflicting definition for itemtype")
288 else:
289 results["itemtype"] = itemType
290 return results
292 def __init__(
293 self,
294 doc,
295 keytype=None,
296 itemtype=None,
297 default=None,
298 optional=False,
299 dictCheck=None,
300 keyCheck=None,
301 itemCheck=None,
302 deprecated=None,
303 ):
304 source = getStackFrame()
305 self._setup(
306 doc=doc,
307 dtype=Dict,
308 default=default,
309 check=None,
310 optional=optional,
311 source=source,
312 deprecated=deprecated,
313 )
314 if keytype is None: 314 ↛ 315line 314 didn't jump to line 315 because the condition on line 314 was never true
315 raise ValueError(
316 "keytype must either be supplied as an argument or as a type argument to the class"
317 )
318 if keytype not in self.supportedTypes: 318 ↛ 319line 318 didn't jump to line 319 because the condition on line 318 was never true
319 raise ValueError(f"'keytype' {_typeStr(keytype)} is not a supported type")
320 elif itemtype is not None and itemtype not in self.supportedTypes: 320 ↛ 321line 320 didn't jump to line 321 because the condition on line 320 was never true
321 raise ValueError(f"'itemtype' {_typeStr(itemtype)} is not a supported type")
323 check_errors = []
324 for name, check in (("dictCheck", dictCheck), ("keyCheck", keyCheck), ("itemCheck", itemCheck)):
325 if check is not None and not callable(check): 325 ↛ 326line 325 didn't jump to line 326 because the condition on line 325 was never true
326 check_errors.append(name)
327 if check_errors: 327 ↛ 328line 327 didn't jump to line 328 because the condition on line 327 was never true
328 raise ValueError(f"{', '.join(check_errors)} must be callable")
330 self.keytype = keytype
331 self.itemtype = itemtype
332 self.dictCheck = dictCheck
333 self.keyCheck = keyCheck
334 self.itemCheck = itemCheck
336 def validate(self, instance):
337 """Validate the field's value (for internal use only).
339 Parameters
340 ----------
341 instance : `lsst.pex.config.Config`
342 The configuration that contains this field.
344 Raises
345 ------
346 lsst.pex.config.FieldValidationError
347 Raised if validation fails for this field (see *Notes*).
349 Notes
350 -----
351 This method validates values according to the following criteria:
353 - A non-optional field is not `None`.
354 - If a value is not `None`, it must pass the `ConfigField.dictCheck`
355 user callback function.
357 Individual key and item checks by the ``keyCheck`` and ``itemCheck``
358 user callback functions are done immediately when the value is set on a
359 key. Those checks are not repeated by this method.
360 """
361 Field.validate(self, instance)
362 value = self.__get__(instance)
363 if value is not None and self.dictCheck is not None and not self.dictCheck(value):
364 msg = f"{value} is not a valid value"
365 raise FieldValidationError(self, instance, msg)
367 def __set__(
368 self,
369 instance: Config,
370 value: Mapping[KeyTypeVar, ItemTypeVar] | None,
371 at: Any = None,
372 label: str = "assignment",
373 ) -> None:
374 if instance._frozen:
375 msg = f"Cannot modify a frozen Config. Attempting to set field to value {value}"
376 raise FieldValidationError(self, instance, msg)
378 if at is None:
379 at = getCallStack()
380 if value is not None:
381 value = self.DictClass(instance, self, value, at=at, label=label)
382 else:
383 history = instance._history.setdefault(self.name, [])
384 history.append((value, at, label))
386 instance._storage[self.name] = value
388 def toDict(self, instance):
389 """Convert this field's key-value pairs into a regular `dict`.
391 Parameters
392 ----------
393 instance : `lsst.pex.config.Config`
394 The configuration that contains this field.
396 Returns
397 -------
398 result : `dict` or `None`
399 If this field has a value of `None`, then this method returns
400 `None`. Otherwise, this method returns the field's value as a
401 regular Python `dict`.
402 """
403 value = self.__get__(instance)
404 return dict(value) if value is not None else None
406 def _compare(self, instance1, instance2, shortcut, rtol, atol, output):
407 """Compare two fields for equality.
409 Used by `lsst.pex.ConfigDictField.compare`.
411 Parameters
412 ----------
413 instance1 : `lsst.pex.config.Config`
414 Left-hand side config instance to compare.
415 instance2 : `lsst.pex.config.Config`
416 Right-hand side config instance to compare.
417 shortcut : `bool`
418 If `True`, this function returns as soon as an inequality if found.
419 rtol : `float`
420 Relative tolerance for floating point comparisons.
421 atol : `float`
422 Absolute tolerance for floating point comparisons.
423 output : callable
424 A callable that takes a string, used (possibly repeatedly) to
425 report inequalities.
427 Returns
428 -------
429 isEqual : bool
430 `True` if the fields are equal, `False` otherwise.
432 Notes
433 -----
434 Floating point comparisons are performed by `numpy.allclose`.
435 """
436 d1 = getattr(instance1, self.name)
437 d2 = getattr(instance2, self.name)
438 name = getComparisonName(
439 _joinNamePath(instance1._name, self.name), _joinNamePath(instance2._name, self.name)
440 )
441 if d1 is None or d2 is None:
442 return compareScalars(name, d1, d2, output=output)
443 if not compareScalars(f"{name} (keys)", set(d1.keys()), set(d2.keys()), output=output):
444 return False
445 equal = True
446 for k, v1 in d1.items():
447 v2 = d2[k]
448 result = compareScalars(
449 f"{name}[{k!r}]", v1, v2, dtype=self.itemtype, rtol=rtol, atol=atol, output=output
450 )
451 if not result and shortcut:
452 return False
453 equal = equal and result
454 return equal