[QI2-1509] Add run_async to qi2_shared

Author: Mythir

qi2_shared/utils.py

@@ -0,0 +1,13 @@
+import asyncio
+import concurrent
+from typing import Any, Coroutine
+
+
+def run_async(async_function: Coroutine[Any, Any, Any]) -> Any:
+    try:
+        _ = asyncio.get_running_loop()
+    except RuntimeError:
+        return asyncio.run(async_function)
+
+    with concurrent.futures.ThreadPoolExecutor() as executor:
+        return executor.submit(asyncio.run, async_function).result()

tests/test_utils.py

@@ -0,0 +1,20 @@
+import asyncio
+
+from qi2_shared.utils import run_async
+
+
+def test_async_run_no_loop() -> None:
+    async def t_coro() -> None:
+        await asyncio.sleep(1)
+
+    run_async(t_coro())
+
+
+def test_async_run_loop() -> None:
+    async def t_coro() -> None:
+        await asyncio.sleep(1)
+
+    async def main() -> None:
+        run_async(t_coro())
+
+    asyncio.run(main())