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

adds trigger_method function to reduce code duplication

Browse Source
This commit is contained in:
Mose Müller 2024-04-16 09:59:39 +02:00
parent cf637d19ae
commit d48ae9f5ad
1 changed files with 43 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

167
src/pydase/client/proxy_loader.py
View File

@ -20,6 +20,36 @@ logger = logging.getLogger(__name__)
class ProxyAttributeError(Exception): ... class ProxyAttributeError(Exception): ...
def trigger_method(
sio_client: socketio.AsyncClient,
loop: asyncio.AbstractEventLoop,
access_path: str,
args: list[Any],
kwargs: dict[str, Any],
) -> Any:
async def async_trigger_method() -> Any:
return await sio_client.call(
"trigger_method",
{
"access_path": access_path,
"args": dump(args),
"kwargs": dump(kwargs),
},
)
result: SerializedObject | None = asyncio.run_coroutine_threadsafe(
async_trigger_method(),
loop=loop,
).result()
if result is not None:
return ProxyLoader.loads_proxy(
serialized_object=result, sio_client=sio_client, loop=loop
)
return None
class ProxyList(list[Any]): class ProxyList(list[Any]):
def __init__( def __init__(
self, self,
@ -57,135 +87,32 @@ class ProxyList(list[Any]):
def append(self, __object: Any) -> None: def append(self, __object: Any) -> None:
full_access_path = f"{self._parent_path}.append" full_access_path = f"{self._parent_path}.append"
async def set_result() -> Any: trigger_method(self._sio, self._loop, full_access_path, [__object], {})
return await self._sio.call(
"trigger_method",
{
"access_path": full_access_path,
"args": dump([__object]),
"kwargs": dump({}),
},
)
result: SerializedObject | None = asyncio.run_coroutine_threadsafe(
set_result(),
loop=self._loop,
).result()
if result is not None:
ProxyLoader.loads_proxy(
serialized_object=result, sio_client=self._sio, loop=self._loop
)
def clear(self) -> None: def clear(self) -> None:
full_access_path = f"{self._parent_path}.clear" full_access_path = f"{self._parent_path}.clear"
async def set_result() -> Any: trigger_method(self._sio, self._loop, full_access_path, [], {})
return await self._sio.call(
"trigger_method",
{
"access_path": full_access_path,
"args": dump([]),
"kwargs": dump({}),
},
)
result: SerializedObject | None = asyncio.run_coroutine_threadsafe(
set_result(),
loop=self._loop,
).result()
if result is not None:
ProxyLoader.loads_proxy(
serialized_object=result, sio_client=self._sio, loop=self._loop
)
def extend(self, __iterable: Iterable[Any]) -> None: def extend(self, __iterable: Iterable[Any]) -> None:
full_access_path = f"{self._parent_path}.extend" full_access_path = f"{self._parent_path}.extend"
async def set_result() -> Any: trigger_method(self._sio, self._loop, full_access_path, [__iterable], {})
return await self._sio.call(
"trigger_method",
{
"access_path": full_access_path,
"args": dump([__iterable]),
"kwargs": dump({}),
},
)
result: SerializedObject | None = asyncio.run_coroutine_threadsafe(
set_result(),
loop=self._loop,
).result()
if result is not None:
ProxyLoader.loads_proxy(
serialized_object=result, sio_client=self._sio, loop=self._loop
)
def insert(self, __index: SupportsIndex, __object: Any) -> None: def insert(self, __index: SupportsIndex, __object: Any) -> None:
full_access_path = f"{self._parent_path}.insert" full_access_path = f"{self._parent_path}.insert"
async def set_result() -> Any: trigger_method(self._sio, self._loop, full_access_path, [__index, __object], {})
return await self._sio.call(
"trigger_method",
{
"access_path": full_access_path,
"args": dump([__index, __object]),
"kwargs": dump({}),
},
)
result: SerializedObject | None = asyncio.run_coroutine_threadsafe(
set_result(),
loop=self._loop,
).result()
if result is not None:
ProxyLoader.loads_proxy(
serialized_object=result, sio_client=self._sio, loop=self._loop
)
def pop(self, __index: SupportsIndex = -1) -> Any: def pop(self, __index: SupportsIndex = -1) -> Any:
full_access_path = f"{self._parent_path}.pop" full_access_path = f"{self._parent_path}.pop"
async def set_result() -> Any: return trigger_method(self._sio, self._loop, full_access_path, [__index], {})
return await self._sio.call(
"trigger_method",
{
"access_path": full_access_path,
"args": dump([__index]),
"kwargs": dump({}),
},
)
result: SerializedObject | None = asyncio.run_coroutine_threadsafe(
set_result(),
loop=self._loop,
).result()
if result is not None:
return ProxyLoader.loads_proxy(
serialized_object=result, sio_client=self._sio, loop=self._loop
)
return None
def remove(self, __value: Any) -> None: def remove(self, __value: Any) -> None:
full_access_path = f"{self._parent_path}.remove" full_access_path = f"{self._parent_path}.remove"
async def set_result() -> Any: trigger_method(self._sio, self._loop, full_access_path, [__value], {})
return await self._sio.call(
"trigger_method",
{
"access_path": full_access_path,
"args": dump([__value]),
"kwargs": dump({}),
},
)
result: SerializedObject | None = asyncio.run_coroutine_threadsafe(
set_result(),
loop=self._loop,
).result()
if result is not None:
ProxyLoader.loads_proxy(
serialized_object=result, sio_client=self._sio, loop=self._loop
)
class ProxyClassMixin: class ProxyClassMixin:
@ -259,21 +186,13 @@ class ProxyClassMixin:
self, attr_name: str, serialized_object: SerializedObject self, attr_name: str, serialized_object: SerializedObject
) -> None: ) -> None:
def method_proxy(*args: Any, **kwargs: Any) -> Any: def method_proxy(*args: Any, **kwargs: Any) -> Any:
async def trigger_method() -> Any: return trigger_method(
return await self._sio.call( self._sio,
"trigger_method", self._loop,
{ serialized_object["full_access_path"],
"access_path": serialized_object["full_access_path"], list(args),
"args": dump(list(args)), kwargs,
"kwargs": dump(kwargs), )
},
)
result = asyncio.run_coroutine_threadsafe(
trigger_method(),
loop=self._loop,
).result()
return loads(result)
dict.__setitem__(self._proxy_methods, attr_name, method_proxy) dict.__setitem__(self._proxy_methods, attr_name, method_proxy)