Add set_attribute_value_setter() method to server

This method allows to set a custom setter function on a per-attribute
basis. This setter function will then be called by
`write_attribute_value()` instead of writing the attribute value
directly. This allows for custom logic to be implemented by the user
(such as checking if the value to be written is within a certain range)
and if not, an appropriate error can be raised from that setter,
preventing the write and notifying the client about the occurred error.

asyncua/server/address_space.py

@@ -40,6 +40,7 @@ class AttributeValue(object):
     def __init__(self, value: ua.DataValue):
         self.value: Optional[ua.DataValue] = value
         self.value_callback: Optional[Callable[[ua.NodeId, ua.AttributeIds], ua.DataValue]] = None
+        self.value_setter: Optional[Callable[["NodeData", ua.AttributeIds, ua.DataValue], None]] = None
         self.datachange_callbacks = {}
 
     def __str__(self) -> str:
@@ -790,7 +791,10 @@ async def write_attribute_value(self, nodeid: ua.NodeId, attr: ua.AttributeIds,
             # Only check datatype if no bad StatusCode is set
             return ua.StatusCode(ua.StatusCodes.BadTypeMismatch)
 
-        attval.value = value
+        if attval.value_setter is not None:
+            attval.value_setter(node, attr, value)
+        else:
+            attval.value = value
         attval.value_callback = None
 
         for k, v in attval.datachange_callbacks.items():
@@ -847,6 +851,21 @@ def set_attribute_value_callback(
 
         return ua.StatusCode()
 
+    def set_attribute_value_setter(
+        self,
+        nodeid: ua.NodeId,
+        attr: ua.AttributeIds,
+        setter: Callable[[NodeData, ua.AttributeIds], ua.DataValue],
+    ) -> ua.StatusCode:
+        node = self._nodes.get(nodeid, None)
+        if node is None:
+            return ua.StatusCode(ua.StatusCodes.BadNodeIdUnknown)
+        attval = node.attributes.get(attr, None)
+        if attval is None:
+            return ua.StatusCode(ua.StatusCodes.BadAttributeIdInvalid)
+        attval.value_setter = setter
+        return ua.StatusCode()
+
     def add_datachange_callback(self, nodeid: ua.NodeId, attr: ua.AttributeIds, callback: Callable) -> Tuple[ua.StatusCode, int]:
         # self.logger.debug("set attr callback: %s %s %s", nodeid, attr, callback)
         if nodeid not in self._nodes:

asyncua/server/internal_server.py

@@ -16,7 +16,7 @@
 from ..common.callback import CallbackService
 from ..common.node import Node
 from .history import HistoryManager
-from .address_space import AddressSpace, AttributeService, ViewService, NodeManagementService, MethodService
+from .address_space import NodeData, AddressSpace, AttributeService, ViewService, NodeManagementService, MethodService
 from .subscription_service import SubscriptionService
 from .standard_address_space import standard_address_space
 from .users import User, UserRole
@@ -353,6 +353,20 @@ def set_attribute_value_callback(
         """
         self.aspace.set_attribute_value_callback(nodeid, attr, callback)
 
+    def set_attribute_value_setter(
+        self,
+        nodeid: ua.NodeId,
+        setter: Callable[[NodeData, ua.AttributeIds, ua.DataValue], None],
+        attr=ua.AttributeIds.Value
+    ) -> None:
+        """
+        Set a setter function for the Attribute. This setter will be called when a new value is set using
+        write_attribute_value() instead of directly writing the value. This is useful, for example, if you want to
+        intercept writes to certain attributes to perform some kind of validation of the value to be written and return
+        appropriate status codes to the client.
+        """
+        self.aspace.set_attribute_value_setter(nodeid, attr, setter)
+
     def read_attribute_value(self, nodeid, attr=ua.AttributeIds.Value):
         """
         directly read datavalue of the Attribute

asyncua/server/server.py

@@ -13,6 +13,7 @@
 from pathlib import Path
 
 from asyncua import ua
+from .address_space import NodeData
 from .binary_server_asyncio import BinaryServer
 from .internal_server import InternalServer
 from .event_generator import EventGenerator
@@ -848,6 +849,20 @@ def set_attribute_value_callback(
         """
         self.iserver.set_attribute_value_callback(nodeid, callback, attr)
 
+    def set_attribute_value_setter(
+        self,
+        nodeid: ua.NodeId,
+        setter: Callable[[NodeData, ua.AttributeIds, ua.DataValue], None],
+        attr=ua.AttributeIds.Value
+    ) -> None:
+        """
+        Set a setter function for the Attribute. This setter will be called when a new value is set using
+        write_attribute_value() instead of directly writing the value. This is useful, for example, if you want to
+        intercept writes to certain attributes to perform some kind of validation of the value to be written and return
+        appropriate status codes to the client.
+        """
+        self.iserver.set_attribute_value_setter(nodeid, setter, attr)
+
     def read_attribute_value(self, nodeid, attr=ua.AttributeIds.Value):
         """
         directly read datavalue of the Attribute

tests/test_server.py

@@ -684,6 +684,35 @@ def callback(nodeid, attr):
     await server.delete_nodes([node])
 
 
+async def test_server_read_set_attribute_value_setter(server: Server):
+    node = await server.get_objects_node().add_variable(0, "0:TestVar", 0, varianttype=ua.VariantType.Int64)
+    dv = server.read_attribute_value(node.nodeid, attr=ua.AttributeIds.Value)
+    assert dv.Value.Value == 0
+
+    def setter(node_data, attr, value):
+        if value.Value.Value > 100:
+            raise ua.uaerrors.BadOutOfRange()
+        else:
+            node_data.attributes[attr].value = value
+
+    server.set_attribute_value_setter(node.nodeid, setter, attr=ua.AttributeIds.Value)
+
+    dv = ua.DataValue(Value=ua.Variant(Value=10, VariantType=ua.VariantType.Int64))
+    await server.write_attribute_value(node.nodeid, dv, attr=ua.AttributeIds.Value)
+    dv = server.read_attribute_value(node.nodeid, attr=ua.AttributeIds.Value)
+    assert dv.Value.Value == 10
+
+    dv = ua.DataValue(Value=ua.Variant(Value=101, VariantType=ua.VariantType.Int64))
+    try:
+        await server.write_attribute_value(node.nodeid, dv, attr=ua.AttributeIds.Value)
+    except Exception as e:
+        assert isinstance(e, ua.uaerrors.BadOutOfRange)
+    dv = server.read_attribute_value(node.nodeid, attr=ua.AttributeIds.Value)
+    assert dv.Value.Value == 10
+
+    await server.delete_nodes([node])
+
+
 @pytest.fixture(scope="function")
 def restore_transport_limits_server(server: Server):
     # Restore limits after test