wrap handling background tasks in an API that hides the complexities

[WolfgangFahl]

The example:
https://github.com/zauberzeug/nicegui/blob/main/examples/opencv_webcam/main.py

does some magic about handling connections and ProcessPoolExecutor. It would be great to have this in a separate class.

Here is a skeleton i started:

"""
Created on 2023-09-12


"""

import asyncio
import concurrent.futures
import signal
from typing import Callable, Any, Union

class BackgroundTaskHandler:
    """
    A class to handle background tasks, especially those that are CPU-intensive 
    or blocking and thus need to be executed outside the main event loop.
    
    Attributes:
        process_pool_executor (concurrent.futures.ProcessPoolExecutor): Executor to 
            run blocking functions in separate processes.
    """

    def __init__(self):
        """Initializes the background task handler."""
        self.process_pool_executor = concurrent.futures.ProcessPoolExecutor()
        signal.signal(signal.SIGINT, self.handle_sigint)

    async def execute_in_background(self, 
                                    blocking_function: Callable[..., Any], 
                                    *args: Any, 
                                    use_process_pool: bool = False) -> Any:
        """
        Executes a function in the background.

        Args:
            blocking_function (Callable[..., Any]): The function to execute.
            *args (Any): Arguments to pass to the function.
            use_process_pool (bool, optional): Whether to use a process pool executor. 
                Defaults to False, which means a thread executor will be used.

        Returns:
            Any: The result returned by the blocking_function.
        """
        loop = asyncio.get_running_loop()
        if use_process_pool:
            return await loop.run_in_executor(self.process_pool_executor, blocking_function, *args)
        else:
            return await loop.run_in_executor(None, blocking_function, *args)

    async def disconnect(self) -> None:
        """
        Disconnects the task or resources. This is a placeholder method 
        to be implemented by the user based on their specific requirements.
        """
        pass

    def handle_sigint(self, signum: signal.Signals, frame: Union[None, Any]) -> None:
        """
        Handles the SIGINT (Ctrl+C) signal to gracefully disconnect and clean up.

        Args:
            signum (signal.Signals): The signal number.
            frame (Union[None, Any]): The interrupted stack frame.
        """
        asyncio.create_task(self.disconnect())
        signal.signal(signal.SIGINT, signal.default_int_handler)

    async def cleanup(self) -> None:
        """
        Cleans up resources and disconnects tasks. This method should be 
        called before the application exits.
        """
        await self.disconnect()
        self.process_pool_executor.shutdown()

[rodja]

We have implemented something similar in RoSys: https://github.com/zauberzeug/rosys/blob/main/rosys/run.py. Can be called with run.io_bound or run.cpu_bound. We could move it into NiceGUI. What do you think @falkoschindler?

[falkoschindler]

Sounds good. Let's create a PR for it.

[falkoschindler]

I'll close this feature request as completed in PR #1656.