Property side-effects with @lt.on_set

src/labthings_fastapi/__init__.py

@@ -28,7 +28,7 @@
 from .thing_slots import thing_slot
 from .thing_server_interface import ThingServerInterface
 from .thing_class_settings import ThingClassSettings
-from .properties import property, setting, DataProperty, DataSetting
+from .properties import property, setting, on_set, DataProperty, DataSetting
 from .actions import action
 from .endpoints import endpoint
 from . import outputs
@@ -54,6 +54,7 @@
     "ThingClassSettings",
     "property",
     "setting",
+    "on_set",
     "DataProperty",
     "DataSetting",
     "action",

src/labthings_fastapi/properties.py

@@ -48,6 +48,7 @@
 from __future__ import annotations
 import builtins
 from collections.abc import Mapping
+from functools import partial
 from types import EllipsisType
 from typing import (
     Annotated,
@@ -460,7 +461,7 @@
         :return: the default value of this property.
         :raises FeatureNotAvailableError: as this must be overridden.
         """
         raise FeatureNotAvailableError(
             f"{obj.name if obj else self.__class__}.{self.name} can't return a "
             f"default, as it's not supported by {self.__class__}."
         )
@@ -478,7 +479,7 @@
         :param obj: the `~lt.Thing` instance we want to reset.
         :raises FeatureNotAvailableError: as only some subclasses implement resetting.
         """
         raise FeatureNotAvailableError(
             f"{obj.name}.{self.name} cannot be reset, as it's not supported by "
             f"{self.__class__}."
         )
@@ -558,7 +559,7 @@
                 ),
             )
             def reset() -> None:
                 self.reset(thing)
 
     def property_affordance(
         self, thing: Owner, path: str | None = None
@@ -714,6 +715,13 @@
         )
         self.readonly = readonly
 
+    on_set_func: Callable[[Owner, Value], Value] | None = None
+    """A function that is called when the property is set.
+
+    This function must return the new value of the property. If it raises
+    an exception, the property's value will not change.
+    """
+
     def instance_get(self, obj: Owner) -> Value:
         """Return the property's value.
 
@@ -743,11 +751,12 @@
         :param value: the new value for the property.
         :param emit_changed_event: whether to emit a changed event.
         """
+        if self.on_set_func:
+            value = self.on_set_func(obj, value)
         if get_validate_properties_on_set(obj.__class__):
             property_info = self.descriptor_info(obj)
-            obj.__dict__[self.name] = property_info.validate(value)
-        else:
-            obj.__dict__[self.name] = value
+            value = property_info.validate(value)
+        obj.__dict__[self.name] = value
         if emit_changed_event:
             self.emit_changed_event(obj, value)
 
@@ -823,6 +832,102 @@
             )
 
 
+def on_set(
+    property_name: str,
+) -> Callable[[Callable[[Owner, Value], Value]], OnSetDescriptor[Owner, Value]]:
+    """Run a function when a data property is set.
+
+    This decorator causes a method to be called whenever a property
+    is set. The method must return the value (and may modify it), but
+    is not responsible for "remembering" the value: that's done by
+    the data property.
+
+    If the method raises an exception, the property will not change
+    its value, and the error will propagate.
+
+    Side effects should be brief: they are performed synchronously
+    during HTTP request handling, so should not exceed a fraction
+    of a second.
+
+    :param property_name: the name of the property to which we are
+        attaching a side effect.
+    :return: a descriptor object that will attach the method to the
+        property, once the class is fully defined.
+    """
+
+    def decorator(
+        func: Callable[[Owner, Value], Value],
+    ) -> OnSetDescriptor[Owner, Value]:
+        return OnSetDescriptor(property_name=property_name, func=func)
+
+    return decorator
+
+
+class OnSetDescriptor(Generic[Owner, Value]):
+    """A class to add side effects to data properties."""
+
+    def __init__(
+        self, property_name: str, func: Callable[[Owner, Value], Value]
+    ) -> None:
+        """Initialise an OnSetDescriptor.
+
+        :param property_name: the name of the property we're attaching a side-effect to.
+        :param func: the function to run when the property is set.
+        """
+        super().__init__()
+        self.property_name = property_name
+        self.func = func
+
+    def __set_name__(self, owner: type[Owner], name: str) -> None:
+        """Attach the function to the property.
+
+        ``__set_name__`` is part of the Descriptor protocol, and is where we
+        are notified of the owning class and our name.
+
+        :param owner: the class on which we are defined.
+        :param name: the name to which this descriptor is assigned.
+        :raises AttributeError: if the specified property name is missing,
+            not a data property, assigned to multiple times, or overwritten by
+            this descriptor.
+        """
+        if self.property_name == name:
+            msg = f"On-set function '{name}' overwrites its property: rename it."
+            raise AttributeError(msg)
+        prop = getattr(owner, self.property_name, None)
+        if not isinstance(prop, DataProperty):
+            msg = "On-set functions may only be attached to data properties. "
+            msg += f"'{self.property_name}' is not a data property"
+            raise AttributeError(msg)
+        if prop.on_set_func is not None:
+            raise AttributeError(f"'{self.property_name}.on_set' has already been set.")
+        prop.on_set_func = self.func
+
+    @overload
+    def __get__(self, obj: Owner) -> Callable[[Value], Value]: ...
+
+    @overload
+    def __get__(
+        self, obj: None, type: type[Owner]
+    ) -> Callable[[Owner, Value], Value]: ...
+
+    def __get__(
+        self, obj: Owner | None, type: type[Owner] | None = None
+    ) -> Callable[[Owner, Value], Value] | Callable[[Value], Value]:
+        """Return the function.
+
+        As for regular methods, we return the function if accessed on the class, and
+        a bound version if accessed on an instance.
+
+        :param obj: the instance, if accessed on an instance.
+        :param type: the class, if accessed on a class.
+        :return: the function, or a partial object binding the function to the object.
+        """
+        if obj is None:
+            return self.func
+        else:
+            return partial(self.func, obj)
+
+
 class FunctionalProperty(BaseProperty[Owner, Value], Generic[Owner, Value]):
     """A property that uses a getter and a setter.
 
@@ -860,11 +965,11 @@
             # BaseDescriptor parses __doc__ to generate the title and description.
             self.__doc__ = fget.__doc__
         if self._type is None:
             msg = (
                 f"{fget} does not have a valid type. "
                 "Return type annotations are required for property getters."
             )
             raise MissingTypeError(msg)
         self._fset: Callable[[Owner, Value], None] | None = None  # setter function
         # `_freset` should reset the property to its default value.
         self._freset: (
@@ -899,10 +1004,10 @@
         :param fget: The new getter function.
         :return: this descriptor (i.e. ``self``). This allows use as a decorator.
         """
         self._fget = fget
         self._type = return_type(self._fget)
         self.__doc__ = fget.__doc__
         return self
 
     def setter(self, fset: Callable[[Owner, Value], None]) -> Self:
         r"""Set the setter function of the property.
@@ -974,7 +1079,7 @@
             # Don't return the descriptor if it's named differently.
             # see typing notes in docstring.
             return fset  # type: ignore[return-value]
         return self
 
     def instance_get(self, obj: Owner) -> Value:
         """Get the value of the property.
@@ -997,7 +1102,7 @@
         :raises ReadOnlyPropertyError: if the property cannot be set.
         """
         if self.fset is None:
             raise ReadOnlyPropertyError(f"Property {self.name} of {obj} has no setter.")
         if get_validate_properties_on_set(obj.__class__):
             property_info = self.descriptor_info(obj)
             value = property_info.validate(value)

tests/test_property.py

@@ -691,3 +691,63 @@ def myprop(self) -> int:
             @myprop.resetter
             def myprop(self) -> None:
                 pass
+
+
+def test_on_set():
+    """Test that `on_set` works as expected."""
+
+    class Example(lt.Thing):
+        intprop: int = lt.property(default=0)
+
+        shadow: int = lt.property(default=0)
+
+        @lt.on_set("intprop")
+        def _on_set_intprop(self, val: int) -> int:
+            """A function to run when intprop is set."""
+            if val < 0:
+                raise ValueError("Can't be negative.")
+            self.shadow = val
+            return val
+
+    thing = create_thing_without_server(Example)
+    assert thing.shadow == 0
+    thing.intprop = 42
+    assert thing.shadow == 42
+    with pytest.raises(ValueError, match="Can't be negative"):
+        thing.intprop = -1
+
+
+def test_bad_on_set_definitions():
+    """Test that helpful errors are raise if `on_set` is used incorrectly."""
+    with raises_or_is_caused_by(AttributeError) as excinfo:
+
+        class Example2(lt.Thing):
+            @lt.on_set("missing")
+            def set_missing(self, value):
+                return value
+
+    assert "'missing' is not a data property" in str(excinfo)
+
+    with raises_or_is_caused_by(AttributeError) as excinfo:
+
+        class Example3(lt.Thing):
+            @lt.on_set("myprop")
+            def myprop(self, value):
+                return value
+
+    assert "On-set function 'myprop' overwrites its property" in str(excinfo)
+
+    with raises_or_is_caused_by(AttributeError) as excinfo:
+
+        class Example4(lt.Thing):
+            intprop: int = lt.property(default=0)
+
+            @lt.on_set("intprop")
+            def set_intprop(self, value):
+                return value
+
+            @lt.on_set("intprop")
+            def set_intprop2(self, value):
+                return value
+
+    assert "'intprop.on_set' has already been set" in str(excinfo)