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

delay task setup until called from class instance containing the task

Browse Source
This commit is contained in:
Mose Müller 2024-08-06 11:11:44 +02:00
parent 15322b742d
commit 12c0c9763d
1 changed files with 22 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
src/pydase/task/task.py
View File

@ -43,13 +43,10 @@ class Task(pydase.data_service.data_service.DataService, Generic[R]):
autostart: bool = False, autostart: bool = False,
) -> None: ) -> None:
super().__init__() super().__init__()
if not current_event_loop_exists(): self._autostart = autostart
self._loop = asyncio.new_event_loop()
asyncio.set_event_loop(self._loop)
else:
self._loop = asyncio.get_event_loop()
self._func_name = func.__name__ self._func_name = func.__name__
self._bound_func: Callable[[], Coroutine[None, None, R | None]] | None = None self._bound_func: Callable[[], Coroutine[None, None, R | None]] | None = None
self._set_up = False
if is_bound_method(func): if is_bound_method(func):
self._func = func self._func = func
self._bound_func = func self._bound_func = func
@ -58,8 +55,6 @@ class Task(pydase.data_service.data_service.DataService, Generic[R]):
self._task: asyncio.Task[R | None] | None = None self._task: asyncio.Task[R | None] | None = None
self._status = TaskStatus.NOT_RUNNING self._status = TaskStatus.NOT_RUNNING
self._result: R | None = None self._result: R | None = None
if autostart:
self.start()
@property @property
def status(self) -> TaskStatus: def status(self) -> TaskStatus:
@ -111,8 +106,25 @@ class Task(pydase.data_service.data_service.DataService, Generic[R]):
self._task.cancel() self._task.cancel()
def __get__(self, instance: Any, owner: Any) -> Self: def __get__(self, instance: Any, owner: Any) -> Self:
# need to use this descriptor to bind the function to the instance of the class """Descriptor method used to correctly setup the task.
# containing the function
if instance and self._bound_func is None: This descriptor method is called by the class instance containing the task.
We need to use this descriptor to bind the task function to that class instance.
As the __init__ function is called when a function is decorated with
@pydase.task.task, we should delay some of the setup until this descriptor
function is called.
"""
if instance and not self._set_up:
if not current_event_loop_exists():
self._loop = asyncio.new_event_loop()
asyncio.set_event_loop(self._loop)
else:
self._loop = asyncio.get_event_loop()
self._bound_func = self._func.__get__(instance, owner) self._bound_func = self._func.__get__(instance, owner)
self._set_up = True
if self._autostart:
self.start()
return self return self