chore: updating docs, renaming _root to __root__

This commit is contained in:
Mose Müller 2023-08-02 12:06:19 +02:00
parent 206a831473
commit ae8be562db

View File

@ -30,7 +30,7 @@ class DataService(rpyc.Service):
def __init__(self) -> None: def __init__(self) -> None:
# Keep track of the root object. This helps to filter the emission of # Keep track of the root object. This helps to filter the emission of
# notifications # notifications
self._root: "DataService" = self self.__root__: "DataService" = self
# dictionary to keep track of running tasks # dictionary to keep track of running tasks
self.__tasks: dict[str, Future[None]] = {} self.__tasks: dict[str, Future[None]] = {}
@ -62,12 +62,13 @@ class DataService(rpyc.Service):
callback: Callable[[str | int, Any], None], callback: Callable[[str | int, Any], None],
) -> None: ) -> None:
""" """
Register callback to the DataService instance and all its nested instances. Register callback to a DataService or DataServiceList instance and its nested
instances.
This method recursively traverses all attributes of the DataService `obj` and For a DataService, this method traverses its attributes and recursively adds the
adds the callback to each instance's `_callbacks` set when an attribute is a callback for nested DataService or DataServiceList instances. For a
DataService instance. This ensures any modification of attributes within DataServiceList,
nested instances will trigger the provided callback. the callback is also triggered when an item gets reassigned.
""" """
if isinstance(obj, DataServiceList): if isinstance(obj, DataServiceList):
@ -82,7 +83,7 @@ class DataService(rpyc.Service):
for item in obj_list: for item in obj_list:
if isinstance(item, DataService): if isinstance(item, DataService):
item._callbacks.add(callback) item._callbacks.add(callback)
for attr_name in set(dir(item)) - set(dir(object)) - {"_root"}: for attr_name in set(dir(item)) - set(dir(object)) - {"__root__"}:
attr_value = getattr(item, attr_name) attr_value = getattr(item, attr_name)
if isinstance(attr_value, (DataService, DataServiceList)): if isinstance(attr_value, (DataService, DataServiceList)):
self.__register_recursive_parameter_callback( self.__register_recursive_parameter_callback(
@ -95,24 +96,15 @@ class DataService(rpyc.Service):
parent_path: str, parent_path: str,
) -> None: ) -> None:
""" """
Register callbacks to emit notifications when attributes used in a property Register callbacks to notify when properties or their dependencies change.
getter are changed.
This method iterates over all attributes of the class. For each attribute that This method cycles through all attributes (both class and instance level) of the
is a property, it gets the names of the attributes used inside the property's input `obj`. For each attribute that is a property, it identifies dependencies
getter method. It then creates a callback for each of these dependent used in the getter method and creates a callback for each one.
attributes.
If the dependent attribute is a DataServiceList, the callback is added to the The method is recursive for attributes that are of type DataService or
list. So, if any element in the list is changed, the callback will be triggered DataServiceList. It attaches the callback directly to DataServiceList items or
and a notification will be emitted. propagates it through nested DataService instances.
If the dependent attribute is an instance of DataService, the callback is
registered to all nested DataService instances of this attribute using
`_register_recursive_callback`.
For all other types of attributes, the callback is simply added to the
`_callbacks` set of the instance.
""" """
attrs = obj.__get_class_and_instance_attributes() attrs = obj.__get_class_and_instance_attributes()
@ -143,7 +135,7 @@ class DataService(rpyc.Service):
name=dependent_attr, name=dependent_attr,
value=getattr(obj, dependent_attr), value=getattr(obj, dependent_attr),
) )
if self == obj._root if self == obj.__root__
else None else None
) )
@ -158,7 +150,7 @@ class DataService(rpyc.Service):
name=dependent_attr, name=dependent_attr,
value=getattr(obj, dependent_attr), value=getattr(obj, dependent_attr),
) )
if name == dep and self == obj._root if name == dep and self == obj.__root__
else None else None
) )
# Add to _callbacks # Add to _callbacks
@ -213,7 +205,7 @@ class DataService(rpyc.Service):
name=f"{attr_name}[{index}]", name=f"{attr_name}[{index}]",
value=value, value=value,
) )
if self == self._root if self == self.__root__
else None else None
) )
@ -316,7 +308,7 @@ class DataService(rpyc.Service):
lambda name, value: obj._emit_notification( lambda name, value: obj._emit_notification(
parent_path=parent_path, name=name, value=value parent_path=parent_path, name=name, value=value
) )
if self == obj._root if self == obj.__root__
and not name.startswith("_") # we are only interested in public attributes and not name.startswith("_") # we are only interested in public attributes
and not isinstance( and not isinstance(
getattr(type(obj), name, None), property getattr(type(obj), name, None), property
@ -355,7 +347,7 @@ class DataService(rpyc.Service):
"""Handles registration of callbacks for DataService attributes""" """Handles registration of callbacks for DataService attributes"""
# as the DataService is an attribute of self, change the root object # as the DataService is an attribute of self, change the root object
nested_attr._root = self._root nested_attr.__root__ = self.__root__
new_path = f"{parent_path}.{attr_name}" new_path = f"{parent_path}.{attr_name}"
self._register_DataService_callbacks(nested_attr, new_path) self._register_DataService_callbacks(nested_attr, new_path)
@ -412,12 +404,12 @@ class DataService(rpyc.Service):
If an attribute exists at both the instance and class level,the value from the If an attribute exists at both the instance and class level,the value from the
instance attribute takes precedence. instance attribute takes precedence.
The _root object is removed as this will lead to endless recursion in the for The __root__ object is removed as this will lead to endless recursion in the for
loops. loops.
""" """
attrs = dict(chain(type(self).__dict__.items(), self.__dict__.items())) attrs = dict(chain(type(self).__dict__.items(), self.__dict__.items()))
attrs.pop("_root") attrs.pop("__root__")
return attrs return attrs
def serialize(self, prefix: str = "") -> dict[str, dict[str, Any]]: def serialize(self, prefix: str = "") -> dict[str, dict[str, Any]]: