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:
- You define a coroutine using
async def. - The event loop starts the coroutine.
- When the coroutine hits an
awaitstatement, it pauses and returns control to the event loop. - The event loop picks up another coroutine and runs it.
- 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.
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:
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.
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.
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.
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:
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:
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.
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:
httpxoraiohttpinstead ofrequestsasyncio.sleep()instead oftime.sleep()- Async database drivers like
asyncpg(PostgreSQL) ormotor(MongoDB)
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.
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.
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()ortime.sleep()in async code. Use async equivalents likeaiohttpandasyncio.sleep(). - Always
awaitcoroutines and create tasks early withasyncio.create_task()to enable true concurrency.






Comments
Loading comments…