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

adds endpoint function to trigger async function

Browse Source
This commit is contained in:
Mose Müller 2024-12-02 13:40:56 +01:00
parent 59cc834a81
commit 0d1df4f9e5
1 changed files with 20 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
src/pydase/server/web_server/api/v1/endpoints.py
View File

@ -1,4 +1,4 @@
from typing import Any
from typing import TYPE_CHECKING, Any
import pydase.utils.serialization.deserializer
import pydase.utils.serialization.serializer
@ -7,6 +7,9 @@ from pydase.server.web_server.sio_setup import TriggerMethodDict, UpdateDict
from pydase.utils.helpers import get_object_attr_from_path
from pydase.utils.serialization.types import SerializedObject
if TYPE_CHECKING:
from collections.abc import Awaitable, Callable
loads = pydase.utils.serialization.deserializer.loads
Serializer = pydase.utils.serialization.serializer.Serializer
@ -36,3 +39,19 @@ def trigger_method(state_manager: StateManager, data: TriggerMethodDict) -> Any:
kwargs: dict[str, Any] = loads(serialized_kwargs) if serialized_kwargs else {}
return Serializer.serialize_object(method(*args, **kwargs))
async def trigger_async_method(
state_manager: StateManager, data: TriggerMethodDict
) -> Any:
method: Callable[..., Awaitable[Any]] = get_object_attr_from_path(
state_manager.service, data["access_path"]
)
serialized_args = data.get("args", None)
args = loads(serialized_args) if serialized_args else []
serialized_kwargs = data.get("kwargs", None)
kwargs: dict[str, Any] = loads(serialized_kwargs) if serialized_kwargs else {}
return Serializer.serialize_object(await method(*args, **kwargs))