How do you safely execute blocking code from an async function

WHAT THIS TESTS: This question probes whether you understand that asyncio uses a single event loop per thread and that calling a blocking synchronous function directly inside an async coroutine will freeze the entire loop, stalling all concurrent tasks. It also checks if you know the specific asyncio API that bridges sync and async worlds by moving blocking work to a separate thread.

A GOOD ANSWER COVERS: First, explicitly state that direct calls to blocking code inside an async function block the event loop because they never yield control back to the loop. Second, name the modern function asyncio.to_thread or the underlying loop.run_in_executor mechanism. Third, explain the mechanism: the event loop submits the callable to a ThreadPoolExecutor, runs it in a worker thread, and suspends the awaiting coroutine until the result is ready, allowing other coroutines to run in the meantime. Fourth, add nuance that CPU-intensive calculations may still saturate the GIL in a thread, so a ProcessPoolExecutor can be preferable for heavy math, whereas blocking IO like a synchronous database driver is a classic use case for threads.

COMMON WRONG ANSWERS: Calling the function directly and claiming await will help, since await only works on awaitables. Suggesting asyncio.create_task as a fix, which schedules coroutines concurrently but does not prevent blocking code from hogging the loop. Proposing to make the function async by adding the async keyword without actually using non-blocking IO underneath, which creates a fake async function that still blocks. Confusing multithreading with multiprocessing without explaining the GIL or event loop impact.

LIKELY FOLLOW-UPS: When would you choose ProcessPoolExecutor over ThreadPoolExecutor? How does FastAPI handle this under the hood with dependency injection and path operations? What is the default number of worker threads in the executor and how do you configure it? Can you cancel a task running in a thread? How do you propagate contextvars into the thread?

ONE CONCRETE EXAMPLE: Imagine a FastAPI endpoint that must use a legacy synchronous SQLAlchemy session to fetch a user. Writing user = session.get(User, user_id) inside an async def route would block the event loop for every request. Instead, you write user = await asyncio.to_thread(session.get, User, user_id). The session.get call runs in a thread from the default ThreadPoolExecutor, the endpoint coroutine yields control, and the event loop continues serving other requests until the database result returns.

Source: docs.python.org