async/await and asyncio

async def defines a coroutine function. Calling it runs none of its body; it returns a coroutine object, which runs only when it is awaited or wrapped in a task. A coroutine that is never awaited never runs, and when the forgotten object is thrown away Python only emits a RuntimeWarning: "coroutine ... was never awaited".

asyncio.run(main()) starts an event loop on the current thread and runs main to completion. The loop runs one task at a time. A task keeps the thread until it reaches an await on something that is not ready yet — a sleep, a network read — and then the loop switches to another task that can make progress. No preemption, no parallelism: concurrency comes entirely from tasks waiting at the same time.

Concurrent, or just one after another

await fetch_a() followed by await fetch_b() is sequential: b does not start until a has finished. To overlap them, the second must be running before you wait on the first:

• asyncio.create_task(coro) schedules a coroutine to run as soon as the loop gets a chance, and returns a Task you can await later;
• await asyncio.gather(a(), b()) runs them together and returns the results in argument order. By default the first exception is raised to whoever awaits the gather, and the other awaitables are not cancelled: they keep running. With return_exceptions=True, exceptions come back as values in the result list instead;
• async with asyncio.TaskGroup() as tg: (Python 3.11+) waits for every task created in it. If one fails, the others are cancelled, and the failures are raised together as an ExceptionGroup, which except* can pick apart.

1import asyncio2import warnings3 4 5def run():6    log = []7 8    async def worker(name):9        for step in range(2):10            log.append(f"{name}{step}")11            await asyncio.sleep(0)  # give the loop a turn12        return name.upper()13 14    async def main():15        never_awaited = worker("x")16        kind = type(never_awaited).__name__17        with warnings.catch_warnings(record=True) as caught:18            warnings.simplefilter("always")19            del never_awaited  # thrown away without being awaited20        warning = [f"{w.category.__name__}: {w.message}" for w in caught]21 22        await worker("a")23        await worker("b")24        log.append("|")25        results = await asyncio.gather(worker("c"), worker("d"))26        return kind, warning, results27 28    kind, warning, results = asyncio.run(main())29    return {"unawaited_call_returns": kind, "warning": warning, "log": log,30            "gather_results": results}

Never block the loop

Because everything shares one thread, a call that blocks — time.sleep, a synchronous HTTP or database client, a long CPU loop — stops every task until it returns. Use the async versions (await asyncio.sleep, async drivers), or push blocking work to a thread with await asyncio.to_thread(fn, ...).

1import asyncio2import time3 4 5def run():6    async def ticker(log):7        await asyncio.sleep(0)8        log.append("tick")9 10    async def blocking(log):11        tick = asyncio.create_task(ticker(log))12        log.append("start")13        time.sleep(0.05)           # holds the thread14        log.append("end")15        await tick16 17    async def cooperative(log):18        tick = asyncio.create_task(ticker(log))19        log.append("start")20        await asyncio.sleep(0.05)  # lets other tasks run21        log.append("end")22        await tick23 24    with_blocking, with_await = [], []25    asyncio.run(blocking(with_blocking))26    asyncio.run(cooperative(with_await))27    return {"time_sleep": with_blocking, "asyncio_sleep": with_await}

Timeouts and cancellation

asyncio.wait_for(coro, seconds), or the async with asyncio.timeout(seconds): block, cancels the work when time runs out and raises TimeoutError (since 3.11, asyncio.TimeoutError is the built-in one). Cancelling a task raises CancelledError inside it, at the await where it is paused, so its finally blocks run. Catch CancelledError only to clean up, and re-raise it. task.cancel() only requests cancellation; the cleanup happens once the task runs again, so await the task (or gather it) when you need the cleanup finished.

The example below waits on events that are never set instead of long sleeps, so what times out does not depend on the machine's speed. It also puts TaskGroup and gather side by side when one task fails.

1import asyncio2 3 4def run():5    log = []6 7    async def main():8        never = asyncio.Event()  # nobody sets it: waiting on it never ends9 10        async def slow():11            try:12                await never.wait()13                log.append("slow finished")14            finally:15                log.append("slow cleanup ran")16 17        async def sibling():18            try:19                await never.wait()20            except asyncio.CancelledError:21                log.append("sibling cancelled")22                raise23 24        async def failing():25            await asyncio.sleep(0)26            raise ValueError("bad input")27 28        try:29            await asyncio.wait_for(slow(), timeout=0.05)30        except TimeoutError:31            log.append("TimeoutError raised")32 33        try:34            async with asyncio.TaskGroup() as tg:35                tg.create_task(sibling())36                tg.create_task(failing())37        except* ValueError as group:38            messages = [str(e) for e in group.exceptions]39            log.append(f"{type(group).__name__}: {messages}")40 41        go = asyncio.Event()42 43        async def patient():44            await go.wait()45            return "patient finished"46 47        waiting = asyncio.create_task(patient())48        try:49            await asyncio.gather(waiting, failing())50        except ValueError as e:51            log.append(f"gather raised {e!r}; patient cancelled: {waiting.cancelled()}")52        go.set()53        log.append(await waiting)54 55        values = await asyncio.gather(asyncio.sleep(0, "fine"), failing(),56                                      return_exceptions=True)57        log.append([repr(v) for v in values])58 59    asyncio.run(main())60    return {"log": log, "same_timeout_error": asyncio.TimeoutError is TimeoutError}

Common mistakes

• Calling an async function without await. You get a coroutine object, the work never happens, and only a warning says so.
• Awaiting coroutines one by one in a loop and expecting them to run concurrently. Create tasks first, or use gather or a TaskGroup.
• Calling time.sleep or a synchronous client inside a coroutine. The whole program stalls, not just that task.
• Swallowing CancelledError, so a task that was asked to stop keeps running and timeouts quietly stop working.

What interviewers ask

• How is asyncio different from threads? One thread and cooperative switching at await points, against preemptive switching between OS threads; asyncio scales to many waiting tasks cheaply, but any blocking call stalls everything.
• What does gather return, and what happens when one coroutine fails? Results in argument order; by default the first exception is raised to the awaiting code (the others keep running), and with return_exceptions=True exceptions are returned as values instead.
• How do you add a timeout to an async operation? asyncio.wait_for or asyncio.timeout; the work is cancelled, its finally blocks run, and TimeoutError is raised.