Adding asyncio task management via app methods

[alexveden]

My proposal is to add methods for launching / getting / stopping aio.tasks via Textual app methods, instead of asyncio.create_task().

AIO tasks usage pattern is typically long-running procedure in the background, which may emit events, run processes or listen sockets.

Here are main issues I faced with handling Textual tasks:

1. When the task raised an unhandled exception, it's not propagated to the application. (It's an asyncio feature, but it would be great to raise a full app exception when a task fails). However, the exception message is printed in textual dev console, which is good. This issue leads to silenced bugs, when app still running, but background tasks are dead.

2. When the app is closed, the running task doesn't receive any notifications about the shutting down app.

import asyncio
from textual.app import App, ComposeResult
from textual.widgets import Header, Footer

class TaskApp(App):
    
    async def die_task(self):
        # This task silently raises, no effect on app
        raise RuntimeError('ooops')

    async def infinite_task(self):
        try:
            while True:
                await asyncio.sleep(0.1)
        finally:
            # Expected to trigger after program shutdown (asyncio.CancelledError)
            print('Exception')
            self.log('exception')

    def compose(self) -> ComposeResult:
        yield Header()
        yield Footer()
    
    def on_mount(self):
        asyncio.create_task(self.die_task())
        asyncio.create_task(self.infinite_task())


if __name__ == "__main__":
    app = TaskApp()
    app.run()

Rationale

I believe it's common approach for the sake of having TUI responsive to launch long-running functions as an aio.tasks. All issues above can be resolved manually, however it would be nice to have shortcuts for task management in the Textual App.

Proposed API:

def app.task_create(coroutine,  id = None | str, cancel_on_shutdown = True, result_callback = None, remove_when_done=True) -> asyncio.Task:
""""
- Creates new async io task
- Propagates unhandled exceptions of the task
- To be cancelled on textual app shutdown
- Forbids task creation if the same task with id is still running
- Calls callback when task is done
- When id is None, allows to run generic tasks (with duplicated id=None)
- if remove_when_done - removes the task from the internal lists when task.done() (no exception), otherwise user has to manually delete it
""""

def app.task_get(id: str) -> asyncio.Task:
"""
Returns asyncio.task object: https://docs.python.org/3/library/asyncio-task.html#task-object
or  
None if no such `id` has been created via app.task_create()
"""

def app.task_remove(id: str, force_stop=False) -> asyncio.Task:
"""
Remove task if it was created with remove_when_done=False, if it still running and force_stop===True -> do task.cancel() and then remove
"""

Obviously, the app under the hood should check if scheduled tasks are done, raise their unhandled exceptions, run callbacks, and cancel running tasks when the app is shutting down.

Let me know what are your thoughts on this.
Thanks

To be added by @willmcgugan

[willmcgugan]

We do have plans to introduce long running task management, as it is a common requirement. We also want to add a little syntactical sugar, possibly through a decorator.

The underlaying API may look something like you have proposed...