Sync/Async Bridging
Real codebases mix blocking libraries with async frameworks. Bridge with asyncio.to_thread (3.9+), loop.run_in_executor, and asyncio.run only at entrypoints - never call asyncio.run from inside a running loop.
Recipe
import asyncio
def blocking_io() -> str:
return "done"
async def main() -> None:
result = await asyncio.to_thread(blocking_io)
print(result)When to reach for this:
- Legacy sync DB/HTTP SDK in FastAPI route
- CPU-light blocking file or crypto calls
- Invoking async from sync CLI wrapper (careful)
- Gradual migration from Flask to asyncio
- Third-party library with no async support
Working Example
import asyncio
import time
from concurrent.futures import ThreadPoolExecutor
def slow_sync(n: int) -> int:
time.sleep(0.05)
return n * 2
async def via_to_thread() -> int:
return await asyncio.to_thread(slow_sync, 21)
async def via_executor() -> int:
loop = asyncio.get_running_loop()
with ThreadPoolExecutor(max_workers=2) as pool:
return await loop.run_in_executor(pool, slow_sync, 10)
async def main() -> None:
print(await via_to_thread())
print(await via_executor())
asyncio.run(main())What this demonstrates:
to_threadschedules callable on default thread poolrun_in_executoraccepts custom executor for isolation- Blocking work runs off event loop thread
- Loop stays responsive for other coroutines
Deep Dive
Direction Matrix
| From -> To | Tool |
|---|---|
| async -> sync block | to_thread / executor |
| sync -> async entry | asyncio.run(main()) once |
| sync -> async mid-flight | asyncio.run_coroutine_threadsafe (advanced) |
CPU-bound
- Prefer
ProcessPoolExecutorfor CPU-heavy Python to_threaddoes not bypass GIL for compute
Gotchas
- asyncio.run inside running loop - RuntimeError. Fix: await coroutine directly or use threadsafe bridge.
- to_thread flood - thousands of threads queued. Fix: semaphore + bounded pool.
- Blocking ORM in every request - thread pool becomes bottleneck. Fix: native async driver.
- Sharing ORM sessions across threads - unsafe. Fix: session per thread/task.
- Nested asyncio.run in tests - use pytest-asyncio single loop fixture.
Alternatives
| Alternative | Use When | Don't Use When |
|---|---|---|
| Native async client | Available for dependency | N/A |
| Sync framework (WSGI) | Entire app blocking | Need high concurrent I/O |
| anyio | Portable async abstraction | stdlib-only policy |
FAQs
to_thread vs run_in_executor?
to_thread uses default executor - simpler; custom executor when isolation needed.
How many thread workers?
Default ~ min(32, cpu+4) - lower if tasks block long to avoid pile-up.
Call async from sync REPL?
asyncio.run(expr()) in fresh shell only - not from inside async Jupyter without nest_asyncio (avoid in prod).
FastAPI def vs async def?
async def for await; sync def runs in thread pool automatically - still prefer true async I/O.
Process pool from async?
await loop.run_in_executor(process_pool, fn) - watch pickling and main guard.
contextvars in threads?
contextvars.copy_context().run to propagate request id into to_thread work.
Django async views?
ORM still largely sync - sync_to_async wrapper with thread sensitivity flags.
When not to bridge?
If >30% route time in executor - migrate to async library or sync worker deployment model.
time.sleep in thread ok?
Yes in to_thread worker; never in coroutine body.
Testing bridges?
Mock slow_sync to increment counter; assert loop responsive via concurrent heartbeat task.
Related
- threading - thread primitives
- concurrent.futures - executors
- Debugging Async Code - blocking detection
- Async Best Practices - when to migrate off bridge
Stack versions: This page was written for Python 3.14.0 (stable 3.14, maintenance 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, and uv 0.6+.