asyncio & async/await Interview Questions & Answers

asyncio is Python's framework for single-threaded concurrency using an event loop. The event loop is a scheduler that runs many coroutines cooperatively: when one coroutine awaits something (typically I/O), it yields control back to the loop, which runs another ready coroutine while the first waits. No thread is blocked sitting idle.

import asyncio

async def main():
    print("hello")
    await asyncio.sleep(1)    # yields to the loop instead of blocking
    print("world")

asyncio.run(main())           # creates the loop, runs main, then closes it

Crucially this is concurrency, not parallelism — one thread, one core, interleaving tasks at await points. Rule of thumb: asyncio shines when you have many tasks that spend most of their time waiting on I/O.

async def defines a coroutine function — calling it doesn't run the body, it returns a coroutine object that must be awaited or scheduled. await suspends the current coroutine until the awaited awaitable (another coroutine, a Task, or a Future) completes, handing control back to the event loop in the meantime.

import asyncio

async def fetch():
    await asyncio.sleep(1)    # suspension point — loop runs others here
    return "data"

async def main():
    coro = fetch()            # nothing has run yet
    result = await coro       # now it runs; main suspends until it finishes
    print(result)

asyncio.run(main())

You can only await inside an async def. Forgetting to await a coroutine is a common bug — it never runs and you get a "coroutine was never awaited" warning. Think of await as "pause me here and let others run until this is ready."

Threads use preemptive multitasking — the OS can switch threads at any point, so shared state needs locks and context switches are relatively expensive. Coroutines use cooperative multitasking on one thread — switches happen only at explicit await points, so the code between awaits is effectively atomic and switching is cheap.

import asyncio

async def task(name):
    print(f"{name} start")
    await asyncio.sleep(1)        # the ONLY place this can yield
    print(f"{name} done")

async def main():
    await asyncio.gather(task("a"), task("b"))   # thousands are feasible

asyncio.run(main())

Because there's no OS thread per task, you can run tens of thousands of coroutines cheaply, and most data races disappear. The catch: a coroutine that never awaits monopolizes the loop. Threads tolerate blocking code; coroutines do not.

Awaiting coroutines one by one runs them sequentially. To run them concurrently, schedule them together with asyncio.gather (or wrap each in a Task), which lets the loop interleave their await points.

import asyncio

async def fetch(n):
    await asyncio.sleep(1)
    return n * 2

async def main():
    # all three overlap -> ~1 second total, not 3
    results = await asyncio.gather(fetch(1), fetch(2), fetch(3))
    print(results)        # [2, 4, 6] — order matches the arguments

asyncio.run(main())

gather returns results in argument order and, by default, propagates the first exception. asyncio.create_task(coro) schedules a coroutine to start running immediately so it overlaps with later code. The key idea: concurrency comes from scheduling tasks together, not from awaiting them in turn.

Because asyncio runs on one thread, any code that doesn't await — heavy CPU work or blocking synchronous I/O like time.sleep, requests.get, or blocking DB drivers — freezes the entire loop. Every other coroutine stalls until that call returns.

import asyncio, time

async def bad():
    time.sleep(5)              # BLOCKS the whole loop for 5s

async def good():
    await asyncio.sleep(5)     # yields; other tasks keep running

# offload unavoidable blocking/CPU work to a thread or process pool:
async def offloaded():
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(None, time.sleep, 5)   # runs in a thread

Fixes: use async-native libraries (aiohttp, asyncpg), and push CPU-bound or unavoidably-blocking calls into run_in_executor (a thread pool, or a process pool for CPU work). Rule of thumb: inside async code, never call something that blocks without awaiting it.

asyncio wins for high-concurrency I/O-bound workloads — thousands of network calls, web requests, websocket connections, or database queries that spend their time waiting. While one request waits, the loop services others, so a single thread handles huge concurrency cheaply.

import asyncio

async def call_api(i):
    await asyncio.sleep(0.5)        # stand-in for a network round trip
    return i

async def main():
    # 1000 "requests" overlap on one thread in ~0.5s of wall time
    results = await asyncio.gather(*(call_api(i) for i in range(1000)))
    print(len(results))             # 1000

asyncio.run(main())

It does not help CPU-bound work — that needs multiprocessing for real parallelism. And for a handful of blocking calls, plain threads are often simpler. Reach for asyncio when concurrency is high and the bottleneck is waiting on I/O.

Awaiting a coroutine directly runs it to completion before the next line — that is sequential. asyncio.create_task(coro) schedules the coroutine on the loop immediately and returns a Task you can await later, so work overlaps.

import asyncio

async def work(n):
    await asyncio.sleep(1)
    return n

async def main():
    await work(1); await work(2)          # sequential -> ~2s

    t1 = asyncio.create_task(work(1))      # both start now
    t2 = asyncio.create_task(work(2))
    await t1; await t2                     # concurrent -> ~1s

asyncio.run(main())

Rule of thumb: a bare await is "do this now and wait"; create_task is "start this in the background and collect it later".

gather runs awaitables concurrently and returns their results in order as a list; if one raises, gather propagates that exception (unless you pass return_exceptions=True). wait returns two sets — (done, pending) of Tasks — and never raises for you; you inspect each Task's result yourself.

results = await asyncio.gather(a(), b(), c())     # [ra, rb, rc], ordered

done, pending = await asyncio.wait(
    tasks, return_when=asyncio.FIRST_COMPLETED)   # set-based, manual

Rule of thumb: reach for gather when you want all results back in order; use wait only when you need fine control like "return when the first finishes".

By default the first exception propagates out of gather as soon as it occurs; the other tasks keep running but their results (and errors) are lost/ignored. With return_exceptions=True, exceptions are returned in the results list alongside normal values instead of being raised.

async def ok():  return 1
async def bad(): raise ValueError("boom")

res = await asyncio.gather(ok(), bad(), return_exceptions=True)
# res == [1, ValueError('boom')]  -> inspect each item

for r in res:
    if isinstance(r, Exception):
        print("failed:", r)

Rule of thumb: use return_exceptions=True when you want every task to finish and to handle failures individually rather than aborting on the first.

Run it in a thread so the event loop stays free. asyncio.to_thread(fn, *args) (3.9+) is the simple way; under the hood it uses loop.run_in_executor. The blocking call happens on a worker thread and the coroutine simply awaits the result.

import asyncio, time

def blocking_io():
    time.sleep(2)          # a real blocking call (DB driver, requests, ...)
    return "done"

async def main():
    result = await asyncio.to_thread(blocking_io)   # loop stays responsive
    print(result)

asyncio.run(main())

Rule of thumb: never call time.sleep, requests.get, or other blocking functions directly in a coroutine — wrap them in asyncio.to_thread (I/O) or use multiprocessing (CPU).

Wrap it with asyncio.wait_for(aw, timeout), which cancels the awaitable and raises asyncio.TimeoutError if it runs too long. Python 3.11 added the asyncio.timeout() context manager for guarding a whole block.

try:
    data = await asyncio.wait_for(fetch(), timeout=5)
except asyncio.TimeoutError:
    data = None                      # fell back after 5s

# 3.11+ context-manager form:
async with asyncio.timeout(5):
    data = await fetch()

Rule of thumb: any external call deserves a timeout — without one a single hung request can stall a coroutine forever.

Calling task.cancel() requests cancellation: at the task's next await, asyncio raises asyncio.CancelledError inside it. You can clean up in a try/finally or except CancelledError, but you should re-raise it — swallowing it breaks cancellation semantics.

async def worker():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("cleaning up")
        raise                       # re-raise so cancellation completes

async def main():
    t = asyncio.create_task(worker())
    await asyncio.sleep(0.1)
    t.cancel()
    await asyncio.gather(t, return_exceptions=True)

Rule of thumb: cancellation is cooperative — it only takes effect at an await point, and CancelledError should be re-raised after cleanup.

They are the async versions of with and for. async with drives an async context manager (__aenter__/__aexit__), e.g. acquiring a connection that itself awaits. async for iterates an async generator / async iterator (__anext__), awaiting between items — ideal for streaming I/O.

async def stream(url):
    async with session.get(url) as resp:     # awaits the connection
        async for line in resp.content:       # awaits each chunk
            process(line)

Rule of thumb: use them whenever the setup/teardown or the per-item step involves I/O that must be awaited.

Use an asyncio.Semaphore(n) to cap concurrency. Each coroutine acquires it before the protected work and releases on exit; only n can hold it at once, so the rest wait. This prevents hammering an API or exhausting connections when you fan out thousands of tasks.

sem = asyncio.Semaphore(10)          # at most 10 in flight

async def fetch(url):
    async with sem:                  # blocks if 10 already running
        return await get(url)

results = await asyncio.gather(*(fetch(u) for u in urls))

Rule of thumb: when you gather a huge number of tasks, gate them with a Semaphore so you control the real level of parallelism.

It is an async-aware queue for the producer/consumer pattern within one loop. await queue.put(item) and await queue.get() suspend instead of blocking the thread, letting producers and consumers run as concurrent tasks. task_done()/join() let you wait until all items are processed.

q = asyncio.Queue(maxsize=100)

async def producer():
    for i in range(10):
        await q.put(i)

async def consumer():
    while True:
        item = await q.get()
        ...                          # handle item
        q.task_done()

Rule of thumb: use asyncio.Queue (not queue.Queue) to hand work between coroutines — it cooperates with the event loop instead of blocking it.

asyncio.run() creates a new event loop and is meant to be the single top-level entry point. Calling it again while a loop is already running — common inside Jupyter notebooks or another async framework — raises RuntimeError. From inside a coroutine you should simply await, not call asyncio.run again.

async def main(): ...

asyncio.run(main())          # correct: top level, once

async def outer():
    asyncio.run(main())      # WRONG: loop already running
    await main()             # correct: just await

Rule of thumb: call asyncio.run once at the program's entry point; everywhere else inside async code, await the coroutine.