Merge pull request #113 from lsst/tickets/DM-48074

DM-48074: Introduce keyCheck callback function for DictField and ConfigDictField

doc/changes/DM-48074.feature.rst

@@ -0,0 +1 @@
+Introduced the `keyCheck` callback to `DictField` and `ConfigDictField`, allowing custom key validation during assignment. Added unit tests to ensure functionality.

python/lsst/pex/config/configDictField.py

@@ -77,6 +77,11 @@ def __setitem__(self, k, x, at=None, label="setitem", setHistory=True):
             )
             raise FieldValidationError(self._field, self._config, msg)
 
+        # validate key using keycheck
+        if self._field.keyCheck is not None and not self._field.keyCheck(k):
+            msg = f"Key {k!r} is not a valid key"
+            raise FieldValidationError(self._field, self._config, msg)
+
         if at is None:
             at = getCallStack()
         name = _joinNamePath(self._config._name, self._field.name, k)
@@ -127,6 +132,8 @@ class ConfigDictField(DictField):
         Default is `True`.
     dictCheck : `~collections.abc.Callable` or `None`, optional
         Callable to check a dict.
+    keyCheck : `~collections.abc.Callable` or `None`, optional
+        Callable to check a key.
     itemCheck : `~collections.abc.Callable` or `None`, optional
         Callable to check an item.
     deprecated : None or `str`, optional
@@ -140,7 +147,8 @@ class ConfigDictField(DictField):
 
         - ``keytype`` or ``itemtype`` arguments are not supported types
           (members of `ConfigDictField.supportedTypes`.
-        - ``dictCheck`` or ``itemCheck`` is not a callable function.
+        - ``dictCheck``, ``keyCheck`` or ``itemCheck`` is not a callable
+          function.
 
     See Also
     --------
@@ -172,6 +180,7 @@ def __init__(
         default=None,
         optional=False,
         dictCheck=None,
+        keyCheck=None,
         itemCheck=None,
         deprecated=None,
     ):
@@ -189,14 +198,18 @@ def __init__(
             raise ValueError(f"'keytype' {_typeStr(keytype)} is not a supported type")
         elif not issubclass(itemtype, Config):
             raise ValueError(f"'itemtype' {_typeStr(itemtype)} is not a supported type")
-        if dictCheck is not None and not hasattr(dictCheck, "__call__"):
-            raise ValueError("'dictCheck' must be callable")
-        if itemCheck is not None and not hasattr(itemCheck, "__call__"):
-            raise ValueError("'itemCheck' must be callable")
+
+        check_errors = []
+        for name, check in (("dictCheck", dictCheck), ("keyCheck", keyCheck), ("itemCheck", itemCheck)):
+            if check is not None and not hasattr(check, "__call__"):
+                check_errors.append(name)
+        if check_errors:
+            raise ValueError(f"{', '.join(check_errors)} must be callable")
 
         self.keytype = keytype
         self.itemtype = itemtype
         self.dictCheck = dictCheck
+        self.keyCheck = keyCheck
         self.itemCheck = itemCheck
 
     def rename(self, instance):
@@ -207,6 +220,23 @@ def rename(self, instance):
                 configDict[k]._rename(fullname)
 
     def validate(self, instance):
+        """Validate the field.
+
+        Parameters
+        ----------
+        instance : `lsst.pex.config.Config`
+            The config instance that contains this field.
+
+        Raises
+        ------
+        lsst.pex.config.FieldValidationError
+            Raised if validation fails for this field.
+
+        Notes
+        -----
+        Individual key checks (``keyCheck``) are applied when each key is added
+        and are not re-checked by this method.
+        """
         value = self.__get__(instance)
         if value is not None:
             for k in value:

python/lsst/pex/config/dictField.py

@@ -139,6 +139,11 @@ def __setitem__(
                 )
                 raise FieldValidationError(self._field, self._config, msg)
 
+        # validate key using keycheck
+        if self._field.keyCheck is not None and not self._field.keyCheck(k):
+            msg = f"Key {k!r} is not a valid key"
+            raise FieldValidationError(self._field, self._config, msg)
+
         # validate item using itemcheck
         if self._field.itemCheck is not None and not self._field.itemCheck(x):
             msg = f"Item at key {k!r} is not a valid value: {x}"
@@ -214,6 +219,8 @@ class DictField(Field[Dict[KeyTypeVar, ItemTypeVar]], Generic[KeyTypeVar, ItemTy
         If `True`, the field doesn't need to have a set value.
     dictCheck : callable
         A function that validates the dictionary as a whole.
+    keyCheck : callable
+        A function that validates individual mapping keys.
     itemCheck : callable
         A function that validates individual mapping values.
     deprecated : None or `str`, optional
@@ -289,6 +296,7 @@ def __init__(
         default=None,
         optional=False,
         dictCheck=None,
+        keyCheck=None,
         itemCheck=None,
         deprecated=None,
     ):
@@ -310,14 +318,18 @@ def __init__(
             raise ValueError(f"'keytype' {_typeStr(keytype)} is not a supported type")
         elif itemtype is not None and itemtype not in self.supportedTypes:
             raise ValueError(f"'itemtype' {_typeStr(itemtype)} is not a supported type")
-        if dictCheck is not None and not hasattr(dictCheck, "__call__"):
-            raise ValueError("'dictCheck' must be callable")
-        if itemCheck is not None and not hasattr(itemCheck, "__call__"):
-            raise ValueError("'itemCheck' must be callable")
+
+        check_errors = []
+        for name, check in (("dictCheck", dictCheck), ("keyCheck", keyCheck), ("itemCheck", itemCheck)):
+            if check is not None and not hasattr(check, "__call__"):
+                check_errors.append(name)
+        if check_errors:
+            raise ValueError(f"{', '.join(check_errors)} must be callable")
 
         self.keytype = keytype
         self.itemtype = itemtype
         self.dictCheck = dictCheck
+        self.keyCheck = keyCheck
         self.itemCheck = itemCheck
 
     def validate(self, instance):
@@ -328,23 +340,22 @@ def validate(self, instance):
         instance : `lsst.pex.config.Config`
             The configuration that contains this field.
 
-        Returns
-        -------
-        isValid : `bool`
-            `True` is returned if the field passes validation criteria (see
-            *Notes*). Otherwise `False`.
+        Raises
+        ------
+        lsst.pex.config.FieldValidationError
+            Raised if validation fails for this field (see *Notes*).
 
         Notes
         -----
         This method validates values according to the following criteria:
 
         - A non-optional field is not `None`.
-        - If a value is not `None`, is must pass the `ConfigField.dictCheck`
-          user callback functon.
+        - If a value is not `None`, it must pass the `ConfigField.dictCheck`
+          user callback function.
 
-        Individual item checks by the `ConfigField.itemCheck` user callback
-        function are done immediately when the value is set on a key. Those
-        checks are not repeated by this method.
+        Individual key and item checks by the ``keyCheck`` and ``itemCheck``
+        user callback functions are done immediately when the value is set on a
+        key. Those checks are not repeated by this method.
         """
         Field.validate(self, instance)
         value = self.__get__(instance)

python/lsst/pex/config/history.py

@@ -214,7 +214,7 @@ def format(config, name=None, writeSourceLine=True, prefix="", verbose=False):
             if writeSourceLine:
                 line.append(
                     [
-                        "%s" % ("%s:%d" % (frame.filename, frame.lineno)),
+                        f"{frame.filename}:{frame.lineno}",
                         "FILE",
                     ]
                 )
@@ -252,14 +252,14 @@ def format(config, name=None, writeSourceLine=True, prefix="", verbose=False):
     fullname = f"{config._name}.{name}" if config._name is not None else name
     msg.append(_colorize(re.sub(r"^root\.", "", fullname), "NAME"))
     for value, output in outputs:
-        line = prefix + _colorize("%-*s" % (valueLength, value), "VALUE") + " "
+        line = prefix + _colorize(f"{value:<{valueLength}}", "VALUE") + " "
         for i, vt in enumerate(output):
             if writeSourceLine:
-                vt[0][0] = "%-*s" % (sourceLength, vt[0][0])
+                vt[0][0] = f"{vt[0][0]:<{sourceLength}}"
 
             output[i] = " ".join([_colorize(v, t) for v, t in vt])
 
-        line += ("\n%*s" % (valueLength + 1, "")).join(output)
+        line += f"\n{'':>{valueLength + 1}}".join(output)
         msg.append(line)
 
     return "\n".join(msg)

python/lsst/pex/config/listField.py

@@ -115,16 +115,14 @@ def validateItem(self, i, x):
             `ListField.itemCheck` method.
         """
         if not isinstance(x, self._field.itemtype) and x is not None:
-            msg = "Item at position %d with value %s is of incorrect type %s. Expected %s" % (
-                i,
-                x,
-                _typeStr(x),
-                _typeStr(self._field.itemtype),
+            msg = (
+                f"Item at position {i} with value {x} is of incorrect type {_typeStr(x)}. "
+                f"Expected {_typeStr(self._field.itemtype)}"
             )
             raise FieldValidationError(self._field, self._config, msg)
 
         if self._field.itemCheck is not None and not self._field.itemCheck(x):
-            msg = "Item at position %d is not a valid value: %s" % (i, x)
+            msg = f"Item at position {i} is not a valid value: {x}"
             raise FieldValidationError(self._field, self._config, msg)
 
     def list(self):
@@ -329,15 +327,15 @@ def __init__(
             raise ValueError(f"Unsupported dtype {_typeStr(dtype)}")
         if length is not None:
             if length <= 0:
-                raise ValueError("'length' (%d) must be positive" % length)
+                raise ValueError(f"'length' ({length}) must be positive")
             minLength = None
             maxLength = None
         else:
             if maxLength is not None and maxLength <= 0:
-                raise ValueError("'maxLength' (%d) must be positive" % maxLength)
+                raise ValueError(f"'maxLength' ({maxLength}) must be positive")
             if minLength is not None and maxLength is not None and minLength > maxLength:
                 raise ValueError(
-                    "'maxLength' (%d) must be at least as large as 'minLength' (%d)" % (maxLength, minLength)
+                    f"'maxLength' ({maxLength}) must be at least as large as 'minLength' ({minLength})"
                 )
 
         if listCheck is not None and not hasattr(listCheck, "__call__"):
@@ -412,13 +410,13 @@ def validate(self, instance):
         if value is not None:
             lenValue = len(value)
             if self.length is not None and not lenValue == self.length:
-                msg = "Required list length=%d, got length=%d" % (self.length, lenValue)
+                msg = f"Required list length={self.length}, got length={lenValue}"
                 raise FieldValidationError(self, instance, msg)
             elif self.minLength is not None and lenValue < self.minLength:
-                msg = "Minimum allowed list length=%d, got length=%d" % (self.minLength, lenValue)
+                msg = f"Minimum allowed list length={self.minLength}, got length={lenValue}"
                 raise FieldValidationError(self, instance, msg)
             elif self.maxLength is not None and lenValue > self.maxLength:
-                msg = "Maximum allowed list length=%d, got length=%d" % (self.maxLength, lenValue)
+                msg = f"Maximum allowed list length={self.maxLength}, got length={lenValue}"
                 raise FieldValidationError(self, instance, msg)
             elif self.listCheck is not None and not self.listCheck(value):
                 msg = f"{value} is not a valid value"
@@ -510,8 +508,9 @@ def _compare(self, instance1, instance2, shortcut, rtol, atol, output):
         equal = True
         for n, v1, v2 in zip(range(len(l1)), l1, l2):
             result = compareScalars(
-                "%s[%d]" % (name, n), v1, v2, dtype=self.dtype, rtol=rtol, atol=atol, output=output
+                f"{name}[{n}]", v1, v2, dtype=self.dtype, rtol=rtol, atol=atol, output=output
             )
+
             if not result and shortcut:
                 return False
             equal = equal and result

tests/test_configDictField.py

@@ -54,6 +54,14 @@ class Config3(pexConfig.Config):
     field1 = pexConfig.ConfigDictField(keytype=str, itemtype=pexConfig.Config, default={}, doc="doc")
 
 
+class Config4(pexConfig.Config):
+    """Fourth test config."""
+
+    field1 = pexConfig.ConfigDictField(
+        keytype=str, itemtype=pexConfig.Config, default={}, doc="doc", keyCheck=lambda k: k.islower()
+    )
+
+
 class ConfigDictFieldTest(unittest.TestCase):
     """Test of ConfigDictField."""
 
@@ -78,6 +86,16 @@ class BadItemtype(pexConfig.Config):
         else:
             raise SyntaxError("Unsupported itemtypes should not be allowed")
 
+        try:
+
+            class BadKeyCheck(pexConfig.Config):
+                d = pexConfig.ConfigDictField("...", keytype=str, itemtype=Config1, keyCheck=4)
+
+        except Exception:
+            pass
+        else:
+            raise SyntaxError("Non-callable keyCheck should not be allowed")
+
         try:
 
             class BadItemCheck(pexConfig.Config):
@@ -116,6 +134,14 @@ def testValidate(self):
         c.d1["a"].f = 5
         c.validate()
 
+    def testKeyCheckValidation(self):
+        c = Config4()
+        c.field1["lower"] = pexConfig.Config()
+        with self.assertRaises(pexConfig.FieldValidationError, msg="Key check should fail"):
+            c.field1["UPPER"] = pexConfig.Config()
+            # No need for c.validate() here, as the exception for key check is
+            # raised by the assignment.
+
     def testInPlaceModification(self):
         c = Config2(d1={})
         self.assertRaises(pexConfig.FieldValidationError, c.d1.__setitem__, 1, 0)

tests/test_dictField.py

@@ -38,6 +38,8 @@ class Config1(pexConfig.Config):
     d2 = pexConfig.DictField("d2", keytype=str, itemtype=str, default=None)
     d3 = pexConfig.DictField("d3", keytype=float, itemtype=float, optional=True, itemCheck=lambda x: x > 0)
     d4 = pexConfig.DictField("d4", keytype=str, itemtype=None, default={})
+    d5 = pexConfig.DictField[str, float]("d5", default={}, keyCheck=lambda k: k not in ["k1", "k2"])
+    d6 = pexConfig.DictField[int, str]("d6", default={-2: "v1", 4: "v2"}, keyCheck=lambda k: k % 2 == 0)
 
 
 class DictFieldTest(unittest.TestCase):
@@ -64,6 +66,16 @@ class BadItemtype(pexConfig.Config):
         else:
             raise SyntaxError("Unsupported itemtype DictFields should not be allowed")
 
+        try:
+
+            class BadKeyCheck(pexConfig.Config):
+                d = pexConfig.DictField("...", keytype=int, itemtype=int, keyCheck=4)
+
+        except Exception:
+            pass
+        else:
+            raise SyntaxError("Non-callable keyCheck DictFields should not be allowed")
+
         try:
 
             class BadItemCheck(pexConfig.Config):
@@ -139,6 +151,23 @@ def testValidate(self):
         c.d2 = {"a": "b"}
         c.validate()
 
+    def testKeyCheckValidation(self):
+        c = Config1()
+        c.d5 = {"k3": -1, "k4": 0.25}
+        c.d6 = {6: "v3"}
+
+        with self.assertRaises(
+            pexConfig.FieldValidationError,
+            msg="Key check must reject dictionary assignment with invalid keys",
+        ):
+            c.d5 = {"k1": 1.5, "k2": 2.0}
+
+        with self.assertRaises(
+            pexConfig.FieldValidationError,
+            msg="Key check must reject invalid key addition",
+        ):
+            c.d6[3] = "v4"
+
     def testInPlaceModification(self):
         c = Config1()
         self.assertRaises(pexConfig.FieldValidationError, c.d1.__setitem__, 2, 0)