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

adds try...except blocks around update_value and get_value sio events

Browse Source
This commit is contained in:
Mose Müller 2024-03-27 16:18:59 +01:00
parent 780a2466d3
commit e6251975b8
1 changed files with 15 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -126,20 +126,30 @@ def setup_sio_events(sio: socketio.AsyncServer, state_manager: StateManager) ->
logging.debug("Client [%s] disconnected", click.style(str(sid), fg="cyan")) logging.debug("Client [%s] disconnected", click.style(str(sid), fg="cyan"))
@sio.event @sio.event
async def update_value(sid: str, data: UpdateDict) -> None: async def update_value(sid: str, data: UpdateDict) -> SerializedObject | None: # type: ignore
path = data["access_path"] path = data["access_path"]
# this should probably happen within the following function call -> can also # this should probably happen within the following function call -> can also
# look at the current type of the attribute at "path" # look at the current type of the attribute at "path"
new_value = loads(data["value"]) new_value = loads(data["value"])
return state_manager.set_service_attribute_value_by_path( try:
path=path, value=new_value state_manager.set_service_attribute_value_by_path(
) path=path, value=new_value
)
except Exception as e:
logger.exception(e)
return dump(e)
@sio.event @sio.event
async def get_value(sid: str, access_path: str) -> SerializedObject: async def get_value(sid: str, access_path: str) -> SerializedObject:
return state_manager._data_service_cache.get_value_dict_from_cache(access_path) try:
return state_manager._data_service_cache.get_value_dict_from_cache(
access_path
)
except Exception as e:
logger.exception(e)
return dump(e)
@sio.event @sio.event
async def trigger_method(sid: str, data: TriggerMethodDict) -> Any: async def trigger_method(sid: str, data: TriggerMethodDict) -> Any: