Skip to content
ExponaBLOG
PYTHON BACKEND DEVELOPMENT · DAY 007 / 100

Python async await explained: what developers should know

How asyncio runs many I/O tasks on one thread.

Expona Admin8 min read

Why async await matters in production

When you build web services that talk to databases, APIs, or file systems, you quickly learn that waiting for I/O is the bottleneck, not your code. A single blocked request can waste milliseconds that add up to thousands of wasted seconds across your user base. Python async await explained: what developers should know comes down to one principle: let your thread do useful work while waiting for network packets to arrive, instead of sitting idle.

We've all felt the pain. A service that feels fast in development becomes sluggish under load because each request ties up a worker thread for hundreds of milliseconds. With asyncio, you can handle hundreds or thousands of concurrent I/O operations on a single thread. That's not magic—it's event-driven architecture, and it's essential for modern backends.

The event loop — One thread switches between tasks whenever one awaits I/O

At the heart of Python's asyncio is the event loop. Think of it as a traffic controller standing at an intersection. When a car (coroutine) arrives at a red light (awaits I/O), the controller parks it and moves the next car through the intersection. When the red light turns green (I/O completes), the controller brings that car back into motion.

Here's what actually happens:

  1. You define a coroutine using async def.
  2. The event loop starts the coroutine.
  3. When the coroutine hits an await statement, it pauses and returns control to the event loop.
  4. The event loop picks up another coroutine and runs it.
  5. When the first coroutine's I/O completes, the event loop resumes it.

The critical insight: all of this happens on a single thread. No locks, no race conditions, no threads fighting over shared state. The event loop ensures that only one coroutine runs at any given moment, but it switches between them so fast that they feel concurrent.

python
import asyncio

async def fetch_user(user_id):
    """Simulates a database call."""
    print(f"Fetching user {user_id}...")
    await asyncio.sleep(1)  # Simulates I/O wait
    print(f"Got user {user_id}")
    return {"id": user_id, "name": f"User {user_id}"}

async def main():
    # Start two coroutines. The event loop switches between them.
    task1 = asyncio.create_task(fetch_user(1))
    task2 = asyncio.create_task(fetch_user(2))
    
    # Wait for both to complete
    user1 = await task1
    user2 = await task2
    
    print(f"Results: {user1}, {user2}")

asyncio.run(main())

Run this and you'll see the output like:

python
Fetching user 1...
Fetching user 2...
Got user 1
Got user 2
Results: {'id': 1, 'name': 'User 1'}, {'id': 2, 'name': 'User 2'}

Both coroutines started almost immediately, not one after the other. That's because as soon as each one hit await asyncio.sleep(), control returned to the event loop, which picked up the next task.

Await I/O, not CPU — Async helps with network and database calls. CPU work still blocks

Here's where many developers stumble: async is not a general-purpose performance tool. It only helps when you're waiting on I/O. If you're crunching numbers, async won't speed you up.

When a coroutine is running Python bytecode, the event loop is blocked. It can't switch to another coroutine until the current one hits an await statement. So if you do heavy computation inside a coroutine, you've wasted the benefit of concurrency.

python
import asyncio
import time

async def cpu_bound_work():
    """This blocks the event loop."""
    total = sum(range(100_000_000))  # CPU work
    return total

async def io_bound_work():
    """This yields control back."""
    await asyncio.sleep(1)  # Simulates I/O
    return "done"

async def main():
    start = time.time()
    
    # This will take ~2 seconds, not ~1 second
    # because CPU work blocks the event loop
    task1 = asyncio.create_task(cpu_bound_work())
    task2 = asyncio.create_task(io_bound_work())
    
    result1 = await task1
    result2 = await task2
    
    print(f"Completed in {time.time() - start:.2f} seconds")

asyncio.run(main())

This runs in about 2 seconds, not 1 second, because while cpu_bound_work() is running, the event loop can't switch to handle io_bound_work()'s completion.

The rule is simple: use async for I/O (network requests, database queries, file reads) and CPU-bound operations in separate thread pools or processes.

python
import asyncio
from concurrent.futures import ProcessPoolExecutor

async def cpu_bound_work():
    """Run CPU work in a process pool to avoid blocking the event loop."""
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, sum, range(100_000_000))
    return result

async def main():
    result = await cpu_bound_work()
    print(result)

Run tasks concurrently — gather runs coroutines at the same time

asyncio.gather() is your workhorse for running multiple coroutines concurrently and waiting for all of them to finish. It takes a list of coroutines (or tasks) and returns their results in the same order.

python
import asyncio

async def get_user(user_id):
    """Fetch user from database."""
    await asyncio.sleep(0.5)
    return {"id": user_id, "name": f"User {user_id}"}

async def get_orders(user_id):
    """Fetch orders for a user."""
    await asyncio.sleep(0.3)
    return [{"order_id": 1, "total": 99.99}, {"order_id": 2, "total": 49.99}]

async def main():
    user_id = 42
    
    # Run both queries concurrently
    user, orders = await asyncio.gather(
        get_user(user_id),
        get_orders(user_id)
    )
    
    print(f"User: {user}")
    print(f"Orders: {orders}")

asyncio.run(main())

This completes in about 0.5 seconds (the longer of the two waits), not 0.8 seconds (sum of both waits). That's the power of concurrency.

In a real web service, this pattern appears everywhere. You might fetch a user profile, their preferences, and a list of recommendations in parallel:

python
async def handle_request(user_id: int):
    """Handle an API request by fetching related data concurrently."""
    profile, preferences, recommendations = await asyncio.gather(
        fetch_profile(user_id),
        fetch_preferences(user_id),
        fetch_recommendations(user_id)
    )
    
    return {
        "profile": profile,
        "preferences": preferences,
        "recommendations": recommendations
    }

You can also handle partial failures gracefully with the return_exceptions parameter:

python
results = await asyncio.gather(
    get_user(1),
    get_user(2),
    get_user(3),
    return_exceptions=True
)

# results might contain [user_dict, Exception(...), user_dict]
for result in results:
    if isinstance(result, Exception):
        print(f"Failed: {result}")
    else:
        print(f"Got: {result}")

Do not block the loop — Calling requests or time.sleep in async code freezes every request

This is the most common footgun in async Python. When you use blocking libraries inside an async function, you freeze the entire event loop.

python
import asyncio
import requests  # This is a blocking library
import time

async def bad_fetch():
    """DO NOT DO THIS."""
    response = requests.get("https://api.example.com/data")  # Blocks the event loop!
    return response.json()

async def bad_sleep():
    """DO NOT DO THIS."""
    time.sleep(2)  # Blocks the event loop!
    return "done"

async def main():
    # If bad_fetch() takes 5 seconds, ALL other coroutines wait 5 seconds
    task1 = asyncio.create_task(bad_fetch())
    task2 = asyncio.create_task(bad_fetch())
    
    # This doesn't run concurrently. task2 waits for task1 to finish.
    result1, result2 = await asyncio.gather(task1, task2)

asyncio.run(main())

Both requests.get() and time.sleep() are blocking. They don't yield control to the event loop. The event loop can't switch to another coroutine while they're running, so you lose all the benefits of async.

Use async-compatible alternatives:

  • httpx or aiohttp instead of requests
  • asyncio.sleep() instead of time.sleep()
  • Async database drivers like asyncpg (PostgreSQL) or motor (MongoDB)
python
import asyncio
import aiohttp

async def good_fetch():
    """Use aiohttp for concurrent requests."""
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com/data") as response:
            return await response.json()

async def good_sleep():
    """Use asyncio.sleep() to yield control."""
    await asyncio.sleep(2)
    return "done"

async def main():
    task1 = asyncio.create_task(good_fetch())
    task2 = asyncio.create_task(good_fetch())
    
    # Now this truly runs concurrently
    result1, result2 = await asyncio.gather(task1, task2)

asyncio.run(main())

Common mistakes

Forgetting to await: If you write task = get_data() instead of task = await get_data(), you get a coroutine object, not the result. The function never runs.

python
async def get_data():
    return "data"

async def main():
    task = get_data()  # Wrong: task is a coroutine object
    print(task)  # 
    
    result = await get_data()  # Correct
    print(result)  # "data"

Mixing sync and async libraries: Don't use requests, time.sleep(), or other blocking calls in async code. You'll block the entire event loop and defeat the purpose.

Creating tasks too late: If you call await immediately, the second coroutine doesn't start until the first finishes. Use asyncio.create_task() or asyncio.gather() to start them concurrently.

python
async def main():
    # Wrong: sequential
    result1 = await get_data(1)
    result2 = await get_data(2)
    
    # Correct: concurrent
    result1, result2 = await asyncio.gather(
        get_data(1),
        get_data(2)
    )

Key takeaways

  • Async uses a single-threaded event loop that switches between coroutines when they await I/O, multiplying the number of concurrent operations you can handle.
  • Async only helps with I/O-bound work. CPU-intensive tasks block the event loop and should be offloaded to thread or process pools.
  • asyncio.gather() runs multiple coroutines concurrently and waits for all of them to complete, returning results in order.
  • Never call blocking functions like requests.get() or time.sleep() in async code. Use async equivalents like aiohttp and asyncio.sleep().
  • Always await coroutines and create tasks early with asyncio.create_task() to enable true concurrency.
ShareXLinkedInReddit

Found this useful? Support the writing.

Buy me a coffee
Written by
Expona Admin
Site administrator

Comments

Loading comments…

Leave a comment

Comments are reviewed before they appear. Be kind and stay on topic.

Related articles