Last active
November 18, 2025 01:26
-
-
Save DiTo97/bee863029a8e4a18900a9427ab5d042c to your computer and use it in GitHub Desktop.
A collection of utilities for safely executing asynchronous coroutines
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import asyncio | |
| from typing import Any, Coroutine | |
| class _AsyncThread(threading.Thread): | |
| """helper thread class for running async coroutines in a separate thread""" | |
| def __init__(self, coroutine: Coroutine[Any, Any, Any]): | |
| self.coroutine = coroutine | |
| self.result = None | |
| self.exception = None | |
| super().__init__() | |
| def run(self): | |
| try: | |
| self.result = asyncio.run(self.coroutine) | |
| except Exception as e: | |
| self.exception = e | |
| def run_async_safely(coroutine: Coroutine[Any, Any, Any]) -> Any: | |
| """safely runs an async coroutine, handling existing event loops. | |
| This function detects if there's already a running event loop and uses | |
| a separate thread if needed to avoid the "asyncio.run() cannot be called | |
| from a running event loop" error. This is particularly useful in environments | |
| like Jupyter notebooks, FastAPI applications, or other async frameworks. | |
| Args: | |
| coroutine: The coroutine to run | |
| Returns: | |
| The result of the coroutine | |
| Raises: | |
| Any exception raised by the coroutine | |
| """ | |
| try: | |
| loop = asyncio.get_running_loop() | |
| except RuntimeError: | |
| return asyncio.run(coroutine) | |
| if loop and loop.is_running(): | |
| # There's a running loop, use a separate thread | |
| thread = _AsyncThread(coroutine) | |
| thread.start() | |
| thread.join() | |
| if thread.exception: | |
| raise thread.exception | |
| return thread.result | |
| else: | |
| # The loop exists and is not running | |
| return asyncio.run(coroutine) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment