Skip to content

← Concurrency Without Regret step 11 of 38

Easy Primitives

run_sync: the sync boundary a library should expose

Write the sync-to-async boundary that a library should expose.

def run_sync[T](coro: Coroutine[Any, Any, T]) -> T:
  • Run coro to completion on a brand-new event loop and return its result.
  • If the calling thread already has a running loop, raise RuntimeError(NESTED_MESSAGE) — the exact constant provided in the starter — rather than deadlocking, corrupting the caller’s loop, or silently computing an answer on a second loop that shares nothing with it.

Why this is a real function

Every library that has an async core and a sync convenience wrapper needs it. Get it wrong and the failure lands on your user: they call your innocuous fetch() from inside their request handler and get a RuntimeError whose traceback points into your code. Get it right and the error message tells them what to do.

The check is per thread, not per process. A worker thread has no running loop of its own, so an async caller can always reach a sync API through asyncio.to_thread — test case 5 proves your implementation does not over-refuse.

What the report proves

  • value / outcome / message — the result, or the exception type and text.
  • traceback_ok — a coroutine that raises propagates with its own frame still in the traceback, not re-wrapped.
  • loops_seen / distinct_loops — every call gets its own loop. Caching a loop between calls fails here, and it is the bug that makes the second call in a process mysteriously fail with “bound to a different event loop”.
  • all_closed — the loops are closed on the way out. asyncio.run does that for you, along with finalising async generators and shutting down the default executor; a hand-rolled new_event_loop() + run_until_complete() does not.

The detail that catches people

asyncio.get_running_loop() raises RuntimeError when there is no running loop. That is not a failure to handle, it is the answer to your question — and it is the reason get_event_loop() is the wrong tool here. As of Python 3.14 get_event_loop() also raises when there is no current loop, and the whole policy system is removed in 3.16, so code that reaches for it is on a clock.

One more courtesy: when you refuse to run the coroutine, close() it. An un-awaited coroutine object emits a RuntimeWarning from wherever the garbage collector happens to find it, which is never where the mistake was.

Where the type system earns its keep

Coroutine[Any, Any, T] as the parameter type means run_sync(main) — the missing parentheses — is a type error at the call site, and run_sync(main()) gives the caller a real T rather than an Any. A parameter typed Awaitable[T] would accept a Task or Future too, and neither of those can be run by a loop that does not exist yet; the narrower type is the honest one here.

Loading visualization…