ITQC/pydase
24
0
Fork 0
mirror of https://github.com/tiqi-group/pydase.git synced 2025-06-07 05:50:41 +02:00
Code Issues Packages Projects Releases Wiki Activity

task: using functools to get correct func name

Browse Source
This commit is contained in:
Mose Müller 2024-08-05 16:49:16 +02:00
parent c00cf9a6ff
commit 861e89f37a
2 changed files with 3 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/pydase/task/decorator.py
View File

@ -1,4 +1,5 @@
import asyncio import asyncio
import functools
import logging import logging
from collections.abc import Callable, Coroutine from collections.abc import Callable, Coroutine
from typing import Any, Concatenate, ParamSpec, TypeVar from typing import Any, Concatenate, ParamSpec, TypeVar
@ -17,6 +18,7 @@ def task(
def decorator( def decorator(
func: Callable[Concatenate[Any, P], Coroutine[None, None, R]], func: Callable[Concatenate[Any, P], Coroutine[None, None, R]],
) -> Task[P, R]: ) -> Task[P, R]:
@functools.wraps(func)
async def wrapper(self: Any, *args: P.args, **kwargs: P.kwargs) -> R | None: async def wrapper(self: Any, *args: P.args, **kwargs: P.kwargs) -> R | None:
try: try:
return await func(self, *args, **kwargs) return await func(self, *args, **kwargs)

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

@ -61,7 +61,7 @@ class Task(pydase.DataService, Generic[P, R]):
self._result = task.result() self._result = task.result()
logger.info("Starting task") logger.info("Starting task %s", self._func.__name__)
if inspect.iscoroutinefunction(self._bound_func): if inspect.iscoroutinefunction(self._bound_func):
res: Coroutine[None, None, R] = self._bound_func(*args, **kwargs) res: Coroutine[None, None, R] = self._bound_func(*args, **kwargs)
self._task = asyncio.create_task(res) self._task = asyncio.create_task(res)