Python `async` and `await` Explained

The async and await keywords are the surface syntax of Python's asynchronous programming model. They were introduced in Python 3.5 and have been stable since 3.7, but their simplicity hides a runtime model that is fundamentally different from synchronous Python. async def declares a coroutine function, which does not execute when called but returns a coroutine object that must be scheduled by an event loop.

await suspends a coroutine until an awaitable completes, yielding control to the event loop so other coroutines can run. These two keywords, combined with the asyncio module, give you everything you need to write concurrent I/O code in a single thread. If you have read about the difference between sync and async programming, this article covers the hands-on syntax you need to write async code today.

The async and await keywords are not just decorators on synchronous functions. They change how the function executes, how errors propagate, and how control flows between tasks. This article walks through each piece of the syntax with runnable examples, starting from a single coroutine and building to concurrent task groups with error handling.

async def: defining a coroutine function

An async def function is called a coroutine function. When you call it, Python does not execute the function body. It returns a coroutine object, which represents a computation that can be suspended and resumed.

python

async def greet(name):
    return f"Hello, {name}"
 
result = greet("World")
print(type(result))
print(result)

Calling greet("World") does not run the function body at all. It only builds a coroutine object, which is why print(result) shows a coroutine repr instead of the string "Hello, World":

text

<class 'coroutine'>
<coroutine object greet at 0x...>

The coroutine object is inert. It does nothing until it is awaited or wrapped in a task and scheduled on an event loop. Attempting to call a coroutine function without awaiting it is a common beginner mistake.

Python issues a RuntimeWarning when a coroutine is garbage collected without ever being awaited, which helps catch this.

python

import asyncio
 
async def greet(name):
    return f"Hello, {name}"
 
async def main():
    result = await greet("World")
    print(result)
 
asyncio.run(main())

asyncio.run() creates an event loop, runs main() inside it, and closes the loop. This is the standard way to run async code from a synchronous entry point. Call asyncio.run() exactly once per program, at the top level.

Calling it while an event loop is already running raises a RuntimeError.

await: suspending and resuming

await does three things. It suspends the current coroutine. It tells the event loop that this coroutine is waiting for a specific awaitable to complete.

It lets the event loop switch to another coroutine that is ready to run. When the awaited awaitable completes, the event loop resumes the suspended coroutine right after the await line.

python

import asyncio
 
async def fetch_data(source, delay):
    print(f"Starting {source}")
    await asyncio.sleep(delay)
    print(f"Finished {source}")
    return f"Data from {source}"
 
async def main():
    print("Before await")
    result = await fetch_data("API", 0.1)
    print(f"After await: {result}")
 
asyncio.run(main())

Execution reaches "Before await" before main() ever suspends, then runs fetch_data up to its own await point, and only prints the final line after that coroutine resumes and returns:

text

Before await
Starting API
Finished API
After await: Data from API

The await fetch_data("API", 0.1) line suspends main(). During the 0.1-second sleep, the event loop could run other coroutines if any were scheduled. Since main() is the only coroutine, the loop waits idly.

When the sleep completes, main() resumes, and execution continues to the print statement.

The expression after await must be an awaitable: a coroutine, a Task, or a Future. Awaiting a non-awaitable raises TypeError. Most async libraries return coroutines from their async functions, so you await them directly.

create_task: running coroutines concurrently

await alone runs coroutines sequentially. To run coroutines concurrently, wrap them in tasks with asyncio.create_task(). A task schedules the coroutine on the event loop and returns immediately.

The coroutine runs whenever the event loop has a chance to switch to it.

python

import asyncio
import time
 
async def fetch(source, delay):
    await asyncio.sleep(delay)
    return f"Data from {source}"
 
async def main():
    start = time.perf_counter()
 
    task_a = asyncio.create_task(fetch("A", 0.2))
    task_b = asyncio.create_task(fetch("B", 0.1))
 
    result_a = await task_a
    result_b = await task_b
 
    print(result_a, result_b)
    print(f"Total: {time.perf_counter() - start:.2f}s")
 
asyncio.run(main())

Both tasks started at nearly the same moment, so the total runtime tracks the longer 0.2-second delay rather than the sum of both delays, even though the code awaits task A before task B:

text

Data from A Data from B
Total: 0.20s

Both fetches run concurrently. create_task(fetch("A", 0.2)) schedules task A and returns immediately. create_task(fetch("B", 0.1)) schedules task B and returns immediately.

When main() reaches await task_a, task A may not be done yet, so main() suspends. The event loop runs both tasks. Task B finishes first, but main() is waiting for task A specifically.

When task A finishes, main() resumes. At that point, await task_b returns immediately because task B is already done.

The order of await on tasks matters for when the awaiting coroutine resumes but does not affect when the tasks run. All created tasks run concurrently from the moment they are created.

asyncio.gather: running many coroutines together

asyncio.gather() is the simplest way to run multiple coroutines concurrently and collect their results. Pass it several awaitables, and it returns a future that resolves to a list of results in the same order as the inputs.

python

import asyncio
import time
 
async def fetch(url):
    await asyncio.sleep(0.1)
    return f"Content of {url}"
 
async def main():
    start = time.perf_counter()
 
    results = await asyncio.gather(
        fetch("https://a.example.com"),
        fetch("https://b.example.com"),
        fetch("https://c.example.com"),
    )
 
    print(results)
    print(f"Total: {time.perf_counter() - start:.2f}s")
 
asyncio.run(main())

All three fetches overlap during their 0.1-second sleeps, so the total runtime is close to one fetch's duration instead of three, and the results list keeps the same a, b, c order the coroutines were passed in:

text

['Content of https://a.example.com', 'Content of https://b.example.com', 'Content of https://c.example.com']
Total: 0.10s

gather() starts all the coroutines concurrently and waits for all of them to complete. The results list preserves the input order regardless of completion order. If any coroutine raises an exception, gather() raises that exception immediately and attempts to cancel the remaining coroutines.

To collect exceptions instead of propagating them, pass return_exceptions=True:

python

import asyncio
 
async def may_fail(name, should_fail):
    await asyncio.sleep(0.1)
    if should_fail:
        raise ValueError(f"{name} failed")
    return f"{name} succeeded"
 
async def main():
    results = await asyncio.gather(
        may_fail("A", False),
        may_fail("B", True),
        may_fail("C", False),
        return_exceptions=True,
    )
    for result in results:
        if isinstance(result, Exception):
            print(f"Error: {result}")
        else:
            print(f"Success: {result}")
 
asyncio.run(main())

Even though may_fail("B", True) raises a ValueError, the other two coroutines still run to completion, and the exception shows up as a value in the results list at B's position instead of stopping the whole gather() call:

text

Success: A succeeded
Error: B failed
Success: C succeeded

With return_exceptions=True, exceptions are returned as values in the results list instead of being raised. This pattern is essential when you want all coroutines to complete even if some fail.

Error handling in async code

Exceptions inside a coroutine behave like synchronous exceptions within that coroutine. Use try and except around await calls:

python

import asyncio
 
async def risky_operation():
    await asyncio.sleep(0.1)
    raise ValueError("Something went wrong")
 
async def main():
    try:
        await risky_operation()
    except ValueError as e:
        print(f"Caught: {e}")
 
asyncio.run(main())

Unhandled exceptions in tasks created with create_task() are stored on the task and re-raised when the task is awaited. If a task is never awaited, the exception is silently lost unless you install a custom exception handler:

python

import asyncio
 
async def failing_task():
    await asyncio.sleep(0.1)
    raise RuntimeError("Task failed")
 
async def main():
    task = asyncio.create_task(failing_task())
    await asyncio.sleep(0.2)
 
    if task.exception() is not None:
        print(f"Task failed: {task.exception()}")

task.exception() returns the exception if the task raised one, or None if it completed successfully. Calling it before the task completes raises asyncio.InvalidStateError.

TaskGroup: structured concurrency (Python 3.11+)

Python 3.11 introduced asyncio.TaskGroup, which provides structured concurrency. A task group ensures that when the group exits, all tasks inside it have completed. If any task fails, all remaining tasks in the group are cancelled.

python

import asyncio
 
async def worker(name, delay):
    await asyncio.sleep(delay)
    return f"{name} done"
 
async def main():
    async with asyncio.TaskGroup() as tg:
        task_a = tg.create_task(worker("A", 0.2))
        task_b = tg.create_task(worker("B", 0.1))
 
    print(task_a.result())
    print(task_b.result())
 
asyncio.run(main())

The async with block exits only when all tasks in the group have completed. If worker("A") raised an exception, worker("B") would be cancelled, and the exception would propagate from the async with block. This structured approach prevents the common bug where tasks are orphaned and exceptions are silently lost.

TaskGroup is preferred over raw create_task() for new code because it guarantees cleanup. Use create_task() only when you need a fire-and-forget task that outlives the current scope, and ensure you handle its exceptions explicitly.

asyncio.wait_for: timeouts on async operations

asyncio.wait_for() wraps an awaitable with a timeout. If the awaitable does not complete within the timeout, it is cancelled and TimeoutError is raised.

python

import asyncio
 
async def slow_operation():
    await asyncio.sleep(5)
    return "Done"
 
async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=1)
        print(result)
    except asyncio.TimeoutError:
        print("Operation timed out after 1 second")
 
asyncio.run(main())

wait_for() is essential for any async operation that could hang indefinitely: network requests, database queries, or waiting on external resources. Always set timeouts on I/O operations in production code.

Running synchronous code from async

Not every library is async. When you must call a synchronous function from async code, use loop.run_in_executor() to run it in a thread pool without blocking the event loop:

python

import asyncio
import time
 
def blocking_io():
    time.sleep(0.5)
    return "Result from blocking I/O"
 
async def main():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, blocking_io)
    print(result)
 
asyncio.run(main())

The default executor is a ThreadPoolExecutor, the same one covered in creating and managing Python threads. Passing None uses it, or you can pass a custom executor.

run_in_executor() is the bridge between sync and async worlds. Use it sparingly: every call creates a thread handoff, and too many can degrade performance compared to using async-native libraries.

Practical async pattern summary

Start with asyncio.gather() for simple concurrent operations. Use TaskGroup when you need guaranteed cleanup. Use create_task() directly only when you understand the exception handling implications.

And always use wait_for() with timeouts for any operation that contacts an external service.