Asynchronous I/O with `async` and `await`

The Syntax of Asynchronous Python: async def and await

At the core of asyncio are two keywords that enable asynchronous programming in Python: async and await.

1. async def: This syntax is used to define a coroutine function. When you call a function defined with async def, it doesn't execute its code immediately. Instead, it returns a coroutine object, which is an awaitable object.

2. await: This keyword is used inside a coroutine function to pause its execution until an awaitable object (like another coroutine, a Task, or a Future) completes. While the current coroutine is paused, the event loop is free to run other tasks. You can only use await inside an async def function.

To illustrate, let's use a cooking analogy.

A synchronous cook would put a burger on the grill, wait for it to finish completely, and only then start preparing the salad. An asynchronous cook, however, puts the burger on the grill (await cook_burger()), and while it's cooking (an I/O-bound wait), they immediately start chopping vegetables for the salad (await chop_salad()). The await keyword is the moment the cook pauses one task and switches to another that can be worked on.

For a more formal definition of these keywords and their rules, the following resource is an excellent reference.

Writing Your First Coroutine: From Blocking to Non-Blocking

Let's convert a simple synchronous function into a non-blocking coroutine. This is a fundamental skill. A common I/O-bound operation is simply waiting. In synchronous code, you use time.sleep(), which blocks the entire thread. In asynchronous code, you use asyncio.sleep(), which is a coroutine that tells the event loop to pause the current task and work on something else.

Let's see this in action with a concrete example.

The key insight from that example is that when you await asyncio.sleep(1), you are not just pausing; you are explicitly giving the event loop one second to run other scheduled tasks. Because all three count() coroutines were scheduled concurrently with asyncio.gather(), their sleep times overlap, and the total runtime is determined by the longest path, not the sum of all paths.

A Common Mistake: Sequential await vs. Concurrent Tasks

In the last lesson, you saw an animation showing that simply calling await on coroutines one after another results in sequential, not concurrent, execution. Let's revisit this critical concept with code.

Imagine we wrote the main function from the countasync.py example like this:

async def main():



    # This is sequential, NOT concurrent!
    await count()
    await count()
    await count()

This would run the first count() to completion (taking 2 seconds), then the second, then the third, for a total of 6 seconds. The await keyword effectively says, "Stop everything in this function and wait for this one thing to finish."

To achieve true concurrency, you must first schedule all the coroutines as Tasks on the event loop, allowing the loop to switch between them. The asyncio.gather() function is a convenient utility that does this for you: it takes a series of awaitables, schedules them as tasks, and waits for all of them to complete.

The following video provides an excellent visual explanation of this exact scenario.

Practical Application: Concurrent API Calls

Let's apply these concepts to a scenario you're very familiar with as a backend developer: fetching data from multiple API endpoints.

Synchronously, you would loop through a list of URLs and make requests one by one. If each request takes 1 second, fetching 5 URLs will take 5 seconds.

import requests
import time

def sync_fetch_all(urls):
    """Synchronous version - requests run sequentially"""
    results = []
    for url in urls:



        # Each request blocks until complete before starting the next
        response = requests.get(url)
        results.append({"url": url, "status": response.status_code})
    return results

With asyncio, we can send all requests concurrently and wait for them to complete in parallel. The total time will be roughly the time of the single slowest request. To do this, we need an HTTP client library that supports asyncio, such as aiohttp or httpx.

The following video provides a fantastic, step-by-step walkthrough of refactoring a synchronous requests-based script to a high-performance asynchronous one using aiohttp.

Exercise: Concurrent Data Fetching

Now it's your turn to apply what you've learned.

Write a small Python script with the following components:

1. A coroutine function named fetch_data(id: int) that:

   • Prints a message like Fetching data for {id}...
   • Simulates a random network delay between 0.5 and 2.0 seconds using await asyncio.sleep(random.uniform(0.5, 2.0)).
   • Prints a message like ...Data for {id} fetched.
   • Returns a dictionary, for example: {"id": id, "data": f"Sample data for {id}"}.

2. A main() coroutine that:

   • Uses asyncio.gather() to run fetch_data concurrently for IDs 1, 2, and 3.
   • Prints the results returned from gather.
   • Measures and prints the total execution time.

3. Use asyncio.run(main()) to execute your script.

If you implement this correctly, the total execution time should be around 2 seconds (the maximum delay), not the sum of all delays.

import asyncio
import random
import time

async def fetch_data(id: int) -> dict:
    """A coroutine to simulate fetching data with a delay."""
    print(f"Fetching data for {id}...")
    delay = random.uniform(0.5, 2.0)
    await asyncio.sleep(delay)
    print(f"...Data for {id} fetched in {delay:.2f} seconds.")
    return {"id": id, "data": f"Sample data for {id}"}

async def main():
    """Runs the data fetching coroutines concurrently."""
    start_time = time.perf_counter()
    
    results = await asyncio.gather(
        fetch_data(1),
        fetch_data(2),
        fetch_data(3)
    )
    
    end_time = time.perf_counter()
    
    print("\n--- Results ---")
    for result in results:
        print(result)
        
    print(f"\nTotal execution time: {end_time - start_time:.2f} seconds.")

if __name__ == "__main__":
    asyncio.run(main())

Conclusion

Congratulations! You have now moved beyond the theory and have practical experience writing and running concurrent, I/O-bound code. You've seen how async and await work together to yield control to the event loop, and how tools like asyncio.gather can be used to achieve significant performance improvements.

Key Takeaways:

• async def is used to declare a coroutine function, which returns a coroutine object.
• await pauses a coroutine and yields control to the event loop, but only works inside an async def function.
• You must use non-blocking I/O libraries (e.g., asyncio.sleep, aiohttp) that are asyncio-aware. Using a blocking call like time.sleep or requests.get will freeze the event loop.
• To run operations concurrently, you must schedule them as tasks on the event loop. asyncio.gather is a high-level utility that simplifies this process.
• The async with statement is the standard way to manage asynchronous resources like network connections.

Preview of the Next Lesson:

We've used asyncio.create_task() and asyncio.gather() to run our coroutines. Both of these work with a fundamental asyncio object: the Task. In the next lesson, we will dive deeper into asyncio.Task objects. You'll learn how to create them explicitly, manage their lifecycle, and use them to build more complex and controllable concurrent applications.