Replace current notion of task (async functions) with a proper Task object

[mosmuell]

Is your feature request related to a problem? Please describe. At the moment, every asynchronous function is considered as a task, i.e. start and stop functions will be generated automatically. This poses a couple of problems:

1. Asynchronous functions themselves are not really a thing. They cannot be passed to clients. There is only the notion of a task.
2. As public tasks are always displayed in the frontend, async functions with arguments are currently not possible.
3. There are start_ and stop_ functions for each async function.

Describe the solution you'd like A solution could be a decorator that turns functions into objects with start and stop methods, e.g.:

class Test:
    @TaskDecorator
    def my_task(self, something: str) -> str:
        print("Inside task")
        return something

t = Test().my_task
t.start(something="something")

Alternatively, tasks could really be instances of a task object.

[mosmuell]

A simple implementation of such a decorator is:

import logging
from collections.abc import Callable
from typing import Any, Concatenate, Generic, ParamSpec, TypeVar

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

P = ParamSpec("P")
R = TypeVar("R")

class Task(Generic[P, R]):
    def __init__(self, func: Callable[P, R]) -> None:
        self._func = func

    def start(self, *args: P.args, **kwargs: P.kwargs) -> R:
        logger.info("Starting task")
        return self._func(*args, **kwargs)

    def stop(self) -> None:
        logger.info("Stopping task")

class TaskDecorator(Generic[P, R]):
    def __init__(
        self, func: Callable[Concatenate[Any, P], R], *, autostart: bool
    ) -> None:
        self._func = func

    def __get__(self, obj: Any, obj_type: Any) -> Task[P, R]:
        # need to use this descriptor to bind the function to the instance of the class
        # containing the function

        bound_func = self._func.__get__(obj)
        return Task(bound_func)

def task(
    *,
    autostart: bool = True,
) -> Callable[[Callable[Concatenate[Any, P], R]], TaskDecorator[P, R]]:
    def task_decorator(func: Callable[Concatenate[Any, P], R]) -> TaskDecorator[P, R]:
        return TaskDecorator(func, autostart=autostart)

    return task_decorator

class Test:
    @task()
    def my_task(self, something: str = "") -> str:
        print("Inside task")
        return something