frontends (tools) new aio module for tools regarding asyncio:

A first method lets run an async method from blocking code.

rel 424

libervia/frontends/tools/aio.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python3
+
+# Libervia AsyncIO helper methods
+# Copyright (C) 2009-2023 Jérôme Poisson (goffi@goffi.org)
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import asyncio
+from typing import Any, Coroutine
+
+from libervia.backend.core import log as logging
+
+log = logging.getLogger(__name__)
+background_tasks = set()
+
+
+def _on_task_done(task: asyncio.Future) -> None:
+    """Callback function to execute when a task is done.
+
+    @param task: The completed task.
+        Note: The function removes the task from the tracking set and logs any exceptions
+        that might have occurred.
+    """
+    background_tasks.discard(task)
+    e = task.exception()
+    if e is not None:
+        exc_info = (type(e), e, e.__traceback__)
+        log.error("Task failed:", exc_info=exc_info)
+
+
+def run_async(async_method: Coroutine | asyncio.Future) -> None:
+    """Schedules and tracks an asynchronous method.
+
+    @param async_method: The method to be scheduled for execution.
+        Note: The function keeps a strong reference to the task to prevent it from being
+        garbage-collected. In case of task failure, logs the error.
+        cf. https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task
+    """
+    if isinstance(async_method, asyncio.Future):
+        task = asyncio.ensure_future(async_method)
+    else:
+        task = asyncio.create_task(async_method)
+
+    background_tasks.add(task)
+    task.add_done_callback(_on_task_done)