Skip to Content
Course content

210: Writing Async Functions with async/await

Click on the "Edit" button in the top corner of the screen to edit your slide content.

I've seen this a hundred times: a developer discovers async and await and thinks they've found a magic "go fast" button. They slap async def on every function in their codebase, assuming that Python is now magically running everything in parallel on different CPU cores.

The Myth: "async def" Makes Code Parallel

Here is the hard truth: adding async to a function does not make it run in the background. In fact, if you do this, your code might actually run slower. Look at this snippet:

import asyncio
import time

async def fake_network_call(id):
    print(f"Starting task {id}...")
    # This is the mistake!
    time.sleep(1) 
    print(f"Finished task {id}")

async def main():
    # I'm trying to run three of these "concurrently"
    await asyncio.gather(fake_network_call(1), fake_network_call(2), fake_network_call(3))

asyncio.run(main())

You'd expect this to take about one second because the tasks are "async," right? Wrong. It takes three seconds. Why? Because time.sleep() is a blocking call. It tells the entire OS thread to stop and wait. Since Python's async model runs on a single-threaded event loop, when you call time.sleep(), you aren't just pausing that one function—you're freezing the entire loop. No other task can move forward until that sleep finishes.

The Reality: Cooperative Multitasking

To actually get the benefits of async, you have to yield control back to the event loop. That is what await actually does. It's essentially the function saying, "I'm going to be waiting for the network/disk for a while; feel free to run other things until I'm done."

To fix the example above, we replace the blocking time.sleep() with asyncio.sleep(), which is designed to be awaited:

import asyncio

async def fake_network_call(id):
    print(f"Starting task {id}...")
    # We 'await' here, letting the loop switch to other tasks
    await asyncio.sleep(1) 
    print(f"Finished task {id}")

async def main():
    # Now these actually overlap
    await asyncio.gather(fake_network_call(1), fake_network_call(2), fake_network_call(3))

asyncio.run(main())

Now this runs in roughly one second. I like to think of the event loop as a waiter in a restaurant. A "blocking" waiter would stand at your table and wait for you to decide what to order without moving. An "async" waiter takes your order, then immediately goes to check on three other tables while you're still thinking. They aren't magically in five places at once; they're just managing their downtime efficiently.

Connecting the Dots with Coroutines

When you define a function with async def, you aren't creating a regular function anymore—you're creating a coroutine function. When you call it, it doesn't actually execute the code inside immediately; it returns a coroutine object.

This is why you can't just call an async function from a normal sync function and expect it to work. You have two main options:

  • await the coroutine (which only works inside another async def function).
  • Pass the coroutine to the event loop using asyncio.run() (typically done once at the entry point of your app).

If you forget the await keyword, Python won't throw an error immediately. Instead, it will give you a warning: "RuntimeWarning: coroutine 'fake_network_call' was never awaited". If you see that warning, it means you created the "plan" to do work, but you never actually told the loop to execute it.




📋 Practical Task

Exercise: Building a Concurrent API Status Checker

You need to simulate a tool that checks the health of multiple external services. Instead of checking them one-by-one (which would be painfully slow), you will use asyncio to check them concurrently.

Your Task:

  1. Create an async def check_service(name, delay) function. It should print "Checking [name]...", await asyncio.sleep(delay), and then return a string saying "[name] is UP".
  2. Create an async def main() function that uses asyncio.gather() to run the following checks simultaneously:
    • "Auth-Service" with a 1.5 second delay.
    • "Payment-Gateway" with a 0.5 second delay.
    • "Database-Cluster" with a 2.0 second delay.
  3. Print the final list of results returned by gather().
  4. Run the program using asyncio.run(main()).

Expected Outcome: The total execution time should be approximately 2 seconds (the duration of the longest task), not 4 seconds (the sum of all delays).

Rating
0 0

There are no comments for now.

to be the first to leave a comment.