Add support for critical tasks

[mosmuell]

Is your feature request related to a problem? Please describe. When defining tasks in a service, you cannot shut down the application once that task is done / has crashed. For the latter, the service would still run without doing anything (if it is the critical functionality).

Describe the solution you'd like I am thinking of adding support for critical tasks. Once defined, the service will shutdown completely when (one of) the critical task is done / crashed.

CRITICAL_TASKS = ["critical_task_name1", "critical_task_name2"]  # Critical tasks would be defined by the user in the __init__ function of the DataService class

def task_done_callback(task: asyncio.Task, name: str) -> None:
    """Handles tasks that have finished."""

    # ... [rest of the existing code] ...

    # Check if the task is critical and either completed or crashed
    if name in CRITICAL_TASKS:
        if task.exception() is not None or task.done():
            shutdown_application()

def shutdown_application() -> None:
    """Shuts down the entire application."""
    # logic to gracefully shut down the application
    # e.g.
    asyncio.get_event_loop().stop()

[mosmuell]

I would like to add something like this:

import asyncio
import logging

from pydase import DataService, Server

logger = logging.getLogger(__name__)

class Channel(DataService):
    def __init__(self):
        self._readout_delay = 2.0
        self._value = 0
        self._autostart_tasks = {
            "read_sensor_data": {
                "args": (),  # defaults to ()
                "exit_when_stopped": False,  # defaults to False
                "exit_when_exception_raised": True,  # defaults to False
            }
        }
        super().__init__()

    def _read_from_sensor(self):
        logger.debug("Reading sensor data")
        self._value = 1.0

    @property
    def value(self):
        return self._value

    async def read_sensor_data(self):
        while True:
            self._read_from_sensor()
            await asyncio.sleep(1)

if __name__ == "__main__":
    service = Channel()
    Server(service).run()

The keyword arguments have default values and do not have to be specified. I want to still allow users to use the old way of defining autostart_tasks, namely providing a tuple, not a dictionary.