We can't find the internet
Attempting to reconnect
Something went wrong!
Attempting to reconnect
← Concurrency Without Regret step 11 of 38
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
coroto 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.rundoes that for you, along with finalising async generators and shutting down the default executor; a hand-rollednew_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…
Stuck?
Python reference solution
Sign in to attempt this problem and reveal the reference solution.