ITQC/pydase
24
0
Fork 0
mirror of https://github.com/tiqi-group/pydase.git synced 2025-04-21 00:40:01 +02:00
Code Issues Packages Projects Releases Wiki Activity

prevents users to override nested ProxyClasses in sio client proxy

Browse Source
This commit is contained in:
Mose Müller 2024-03-28 10:09:08 +01:00
parent cc3fdfbb27
commit c02c75aab5
1 changed files with 19 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
src/pydase/client/proxy_class_factory.py
View File

@ -15,10 +15,16 @@ from pydase.utils.serialization.serializer import (
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Callable from collections.abc import Callable
import pydase.components
class ProxyClass(pydase.DataService): class ProxyClass(pydase.DataService):
__sio: socketio.Client __sio: socketio.Client
def __setattr__(self, key, value):
# prevent overriding of proxy attributes
if hasattr(self, key) and isinstance(getattr(self, key), ProxyClass):
raise AttributeError(f"{key} is read-only and cannot be overridden.")
super().__setattr__(key, value)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -28,8 +34,8 @@ class ProxyClassFactory:
def __init__(self, sio_client: socketio.Client) -> None: def __init__(self, sio_client: socketio.Client) -> None:
self.sio_client = sio_client self.sio_client = sio_client
def create_proxy(self, data: SerializedObject) -> "ProxyClass": def create_proxy(self, data: SerializedObject) -> ProxyClass:
proxy: "ProxyClass" = self._deserialize(data) proxy: ProxyClass = self._deserialize(data)
return proxy return proxy
def _deserialize(self, serialized_object: SerializedObject) -> Any: def _deserialize(self, serialized_object: SerializedObject) -> Any:
@ -64,7 +70,7 @@ class ProxyClassFactory:
return None return None
def _deserialize_method(self, serialized_object: SerializedMethod) -> Any: def _deserialize_method(self, serialized_object: SerializedMethod) -> Any:
def method_proxy(self: "ProxyClass", *args: Any, **kwargs: Any) -> Any: def method_proxy(self: ProxyClass, *args: Any, **kwargs: Any) -> Any:
serialized_response = cast( serialized_response = cast(
dict[str, Any], dict[str, Any],
self.__sio.call( self.__sio.call(
@ -82,14 +88,17 @@ class ProxyClassFactory:
def _deserialize_component_type( def _deserialize_component_type(
self, serialized_object: SerializedObject, base_class: type self, serialized_object: SerializedObject, base_class: type
) -> "ProxyClass": ) -> ProxyClass:
def add_prefix_to_last_path_element(s: str, prefix: str) -> str: def add_prefix_to_last_path_element(s: str, prefix: str) -> str:
parts = s.split(".") parts = s.split(".")
parts[-1] = f"{prefix}_{parts[-1]}" parts[-1] = f"{prefix}_{parts[-1]}"
return ".".join(parts) return ".".join(parts)
def create_proxy_class(serialized_object: SerializedObject) -> type: def create_proxy_class(serialized_object: SerializedObject) -> type:
class_bases = (base_class,) class_bases = (
ProxyClass,
base_class,
)
class_attrs: dict[str, Any] = {} class_attrs: dict[str, Any] = {}
# Process and add properties based on the serialized object # Process and add properties based on the serialized object
@ -127,7 +136,7 @@ class ProxyClassFactory:
return create_proxy_class(serialized_object)() return create_proxy_class(serialized_object)()
def _create_attr_property(self, serialized_attr: SerializedObject) -> property: def _create_attr_property(self, serialized_attr: SerializedObject) -> property:
def get(self: "ProxyClass") -> Any: # type: ignore def get(self: ProxyClass) -> Any: # type: ignore
return loads( return loads(
cast( cast(
SerializedObject, SerializedObject,
@ -137,7 +146,7 @@ class ProxyClassFactory:
get.__doc__ = serialized_attr["doc"] get.__doc__ = serialized_attr["doc"]
def set(self: "ProxyClass", value: Any) -> None: # type: ignore def set(self: ProxyClass, value: Any) -> None: # type: ignore
result = cast( result = cast(
SerializedObject | None, SerializedObject | None,
self.__sio.call( self.__sio.call(