I hit this on the Superlinked framework when I added an async client that callers needed to use from sync code. The SDK gets called from notebooks, scripts, and pytest, and each one of these environments treats event loops differently.
If you ship an async library, somebody will call it from a notebook. Somebody else will call it from a synchronous script. A third will call it from inside another async framework that already has a running loop. All three need a working run_sync() helper, and they need it to not blow up.
The naive version is asyncio.run(coro). It works for the script case and it breaks in the other two. Notebooks already have a running loop (Jupyter is async under the hood). Pytest with pytest-asyncio does too. asyncio.run raises RuntimeError: This event loop is already running and now the library is unusable in the most common dev environment.
Here is the version I settled on.
import asyncio
import threading
from concurrent.futures import ThreadPoolExecutor
import nest_asyncio
_local = threading.local()
def run(coro):
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = None
if loop is not None:
# Inside a running loop (notebook, pytest-asyncio, etc.)
nest_asyncio.apply(loop)
try:
return loop.run_until_complete(asyncio.ensure_future(coro))
except RuntimeError as exc:
if "already running" not in str(exc):
raise
# Last resort: run on a fresh loop in a worker thread.
with ThreadPoolExecutor(max_workers=1) as pool:
return pool.submit(asyncio.run, coro).result()
# No running loop. Reuse a per-thread loop so setup cost is paid once per thread.
if not hasattr(_local, "loop"):
_local.loop = asyncio.new_event_loop()
return _local.loop.run_until_complete(coro)Roughly 25 lines, and every single branch is in there for a reason I have hit in production.
The “running loop” branch uses nest_asyncio to make the loop re-entrant. nest_asyncio monkey-patches the loop so a nested run_until_complete does not blow up. It is a hack. It is also the only thing that lets a notebook user write result = library.do_thing() instead of result = await library.do_thing().
The thread fallback is for the cases where nest_asyncio itself fails. Some loops (uvloop in certain configurations, certain test runners) just reject the patch. When that happens the function runs the coroutine on a new loop in a worker thread instead. It is slow but it works in every case I have seen.
The “no loop” branch caches a loop in threading.local. Without the cache, every sync call creates a new loop, which is a few hundred microseconds of garbage. With the cache, repeated calls from the same thread reuse the same loop.
I tried a few times to get away with one branch. Just asyncio.run, or only nest_asyncio, or only the thread executor. Each version broke at least one real use case. The shape above, try the running loop, fall back to the thread, fall back to a per-thread cached loop, is the minimum that handled all of them for me.
nest_asyncio is a required dependency here, not an optional one. It is, again, a hack. It is also the difference between “this library works in notebooks” and “this library does not work in notebooks”, and the second answer was not acceptable.
If your library is going to be called from sync code, ship a run helper like this. The wrapper is small enough to write once. Without it, every caller has to invent their own version, and most of them will get it wrong.
