Python async await: Coroutines and Event Loop Explained

Python's async and await keywords let you write concurrent code that pauses and resumes without blocking the operating system thread. The syntax is compact, but the runtime behavior behind it is often misunderstood. This article explains what python async await actually does, how the event loop drives coroutines, and where the pattern fits in production code.

What async and await Actually Do

async def defines a coroutine function. Calling it returns a coroutine object, not a result. The body does not run until the coroutine is awaited or scheduled on an event loop.

async def fetch_data():
    return "data"

coro = fetch_data()  # no output yet
print(coro)  # <coroutine object fetch_data at 0x...>

await suspends the current coroutine until the awaited object completes. The suspension yields control back to the event loop, which can run other coroutines. This is cooperative multitasking: a coroutine only pauses at await points, never in the middle of synchronous code.

The Event Loop: The Scheduler Behind the Scenes

The event loop is the core of asyncio. It maintains a queue of tasks and a set of callbacks. When a coroutine awaits something that is not ready, the loop registers a callback and moves on to the next ready task. When the awaited operation finishes, the loop resumes the coroutine from the exact point where it paused.

This design works well for I/O-bound work because the loop can overlap waiting time with other computation. It does not speed up CPU-bound code; a long synchronous loop inside a coroutine blocks the entire loop.

Writing Your First Coroutine

A minimal coroutine that performs two sequential awaits shows the basic pattern:

import asyncio

async def step_one():
    await asyncio.sleep(1)
    return 1

async def step_two():
    await asyncio.sleep(1)
    return 2

async def main():
    a = await step_one()
    b = await step_two()
    return a + b

result = asyncio.run(main())
print(result)  # 3

asyncio.run(main()) creates a new event loop, runs the coroutine until it completes, and closes the loop. The two await calls run sequentially because each one waits for the previous coroutine to finish. To overlap them, you need to schedule them as tasks.

Running Multiple Coroutines Concurrently

asyncio.create_task schedules a coroutine to run in the background. The task starts executing as soon as the loop gets a chance, but the current coroutine does not block on it unless you await the task.

async def main():
    task1 = asyncio.create_task(step_one())
    task2 = asyncio.create_task(step_two())
    a = await task1
    b = await task2
    return a + b

Now the two sleep calls overlap because both tasks are scheduled before either is awaited. asyncio.gather provides a higher-level way to await multiple tasks together:

async def main():
    results = await asyncio.gather(step_one(), step_two())
    return sum(results)

gather returns a list of results in the order of the arguments. If any coroutine raises an exception, gather propagates it immediately unless you set return_exceptions=True.

Python 3.11 introduced asyncio.TaskGroup, which gives structured concurrency. Tasks are created within a context manager, and the group waits for all of them before exiting. If one task fails, the group cancels the others.

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(step_one())
        task2 = tg.create_task(step_two())
    return task1.result() + task2.result()

TaskGroup is preferable when you need to guarantee that all tasks finish or are cancelled, especially in error-handling paths.

Common Mistakes That Break async/await

Forgetting to Await a Coroutine

Calling a coroutine without await returns a coroutine object that is never executed. The event loop will emit a RuntimeWarning about a never-awaited coroutine, but the code continues silently.

async def main():
    step_one()  # coroutine created but not awaited
    await asyncio.sleep(1)

Blocking the Event Loop

Using synchronous blocking functions inside a coroutine, such as time.sleep or requests.get, stops the entire loop. Other tasks cannot run until the blocking call returns. Use asyncio.sleep for delays and an async HTTP client like aiohttp or httpx for network requests.

Mixing Sync and Async Without a Bridge

You cannot call an async function from a synchronous function without an event loop. asyncio.run creates a new loop each time, so calling it inside a running loop raises RuntimeError. To call async code from sync code, use asyncio.run at the top level or asyncio.get_event_loop().run_until_complete in legacy code.

When async/await Is the Right Tool

Use python async await when your program spends most of its time waiting on external resources: network responses, database queries, file I/O, or API calls. The concurrency model lets you interleave those waits efficiently with a single thread.

For CPU-bound work, like heavy computation or data processing, async does not help. The Global Interpreter Lock (GIL) still limits parallel execution, and the event loop adds overhead. Use multiprocessing or a thread pool for CPU-bound tasks.

Also consider the complexity cost. Async code is harder to debug because the call stack is split across suspension points. If your application only makes a few sequential I/O calls, plain synchronous code is simpler and easier to maintain. Reserve async for cases where you genuinely need concurrency, such as handling many simultaneous connections or fanning out many independent requests.

A Practical Example: Concurrent HTTP Requests

A typical use case is fetching multiple URLs. Using aiohttp (a common async HTTP library) with asyncio.gather:

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as resp:
        return await resp.text()

async def main():
    urls = ["https://example.com", "https://example.org"]
    async with aiohttp.ClientSession() as session:
        tasks = [asyn.create_task(fetch(session, url) for url in urls]
        pages = await asyncio.gather(*tasks)
    return pages

Note the async with context manager: the session is closed only after all requests complete. The await resp.text() reads the full response body, which is also an I/O operation that yields control.

If you need to limit the number of concurrent requests, use asyncio.Semaphore inside the coroutine:

sem = asyncio.Semaphore(10)

async def fetch_with_limit(session, url):
    async with sem:
        return await fetch(session, url)

The semaphore ensures no more than 10 requests are in flight at once, preventing resource exhaustion on the remote server or local file descriptors.

Handling Cancellation and Timeouts Gracefully

Cancellation is a first-class concept in asyncio. When a task is cancelled, asyncio.CancelledError is raised at the current await point. You can catch it to run cleanup logic, but you should re-raise it after cleanup so the cancellation propagates correctly.

async def worker():
    try:
        await long_operation()
    except asyncio.CancelledError:
        await cleanup()
        raise

For timeouts, use asyn.wait_for or asyncio.timeout (Python 3.11+). asyncio.wait_for raises asyncio.TimeoutError if the awaited operation does not finish within the given seconds.

async def main():
    try:
        result = await asyncio.wait_for(fetch_data(), timeout=2.0)
    except asyncio.TimeoutError:
        print("request timed out")

Timeouts are essential in production to avoid hanging tasks that never complete. Always pair them with cancellation handling so that a timeout actually stops the underlying work.

Choosing Between asyncio, Threads, and Processes

The decision between asyncio, threading, and multiprocessing depends on the nature of the workload and the I/O pattern.

asyncio is ideal when you have hundreds or thousands of connections, such as a web server or a crawler. Threads are useful when you cannot rewrite a blocking library to be async. Processes are the only way to use multiple CPU cores for Python computation.

A common hybrid approach is to run blocking code in a thread pool and await the result from async code using asyncio.to_thread:

import asyncio

def blocking_io():
    return open("large_file.txt").read()

async def main():
    data = await asyncio.to_thread(blocking_io)
    return data

to_thread runs the function in a separate thread and returns a coroutine that completes when the function returns. This keeps the event loop responsive while still using synchronous code where necessary.

The Cost of Async: Debugging and Complexity

Async code introduces a different debugging experience. Stack traces are split across await points, and a single logical operation may involve several coroutines. Tools like asyncio debug mode (PYTHONASYNCIODEBUG=1) can help detect unawaited coroutines and slow callbacks, but they do not eliminate the cognitive load.

Error propagation also differs. If a task fails silently because you never await it, the exception is logged and swallowed. Use TaskGroup or always await or attach a done callback to tasks that might raise.

Before adopting async everywhere, measure whether the added complexity pays off. A synchronous version that performs three sequential HTTP requests might be perfectly fine if the total latency is acceptable. async shines when you need to overlap many waits, not when you have a single slow operation.

When you do need it, python async await with asyncio gives you a robust, standard-library-based concurrency model. The key is to respect the event loop, avoid blocking calls, and structure tasks so that failures are visible and cancellable.