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

uses loads instead of Deserializer, allows for monkey-patching loads and dump

Browse Source
This commit is contained in:
Mose Müller 2024-07-09 16:25:35 +02:00
parent cf0780b2ca
commit e7f9ad799c
1 changed files with 9 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
src/pydase/server/web_server/sio_setup.py
View File

@ -5,15 +5,20 @@ from typing import Any, TypedDict
import click import click
import socketio # type: ignore[import-untyped] import socketio # type: ignore[import-untyped]
import pydase.utils.serialization.deserializer
import pydase.utils.serialization.serializer
from pydase.data_service.data_service_observer import DataServiceObserver from pydase.data_service.data_service_observer import DataServiceObserver
from pydase.data_service.state_manager import StateManager from pydase.data_service.state_manager import StateManager
from pydase.utils.helpers import get_object_attr_from_path from pydase.utils.helpers import get_object_attr_from_path
from pydase.utils.logging import SocketIOHandler from pydase.utils.logging import SocketIOHandler
from pydase.utils.serialization.deserializer import Deserializer from pydase.utils.serialization.serializer import SerializedObject
from pydase.utils.serialization.serializer import SerializedObject, dump
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# These functions can be monkey-patched by other libraries at runtime
loads = pydase.utils.serialization.deserializer.loads
dump = pydase.utils.serialization.serializer.dump
class UpdateDict(TypedDict): class UpdateDict(TypedDict):
""" """
@ -159,8 +164,8 @@ def setup_sio_events(sio: socketio.AsyncServer, state_manager: StateManager) ->
method = get_object_attr_from_path( method = get_object_attr_from_path(
state_manager.service, data["access_path"] state_manager.service, data["access_path"]
) )
args = Deserializer.deserialize(data["args"]) args = loads(data["args"])
kwargs: dict[str, Any] = Deserializer.deserialize(data["kwargs"]) kwargs: dict[str, Any] = loads(data["kwargs"])
return dump(method(*args, **kwargs)) return dump(method(*args, **kwargs))
except Exception as e: except Exception as e:
logger.error(e) logger.error(e)