Modern Python applications often need to handle multiple tasks simultaneously — making API calls, reading files, querying databases, or waiting for network responses. Traditional synchronous code handles these tasks one at a time, wasting valuable CPU cycles while waiting for I/O operations to complete. This is where Python's async/await syntax comes into play.
In this guide, you'll learn what asynchronous programming is, how Python's async/await works under the hood, and how to use it effectively in real-world scenarios.
Understanding the Problem: Synchronous vs Asynchronous
Let's start with a simple example. Imagine you need to fetch data from three different APIs:
import time
import requests
def fetch_data(url):
print(f"Fetching {url}...")
response = requests.get(url)
return response.json()
def main():
start = time.time()
urls = [
"https://jsonplaceholder.typicode.com/posts/1",
"https://jsonplaceholder.typicode.com/posts/2",
"https://jsonplaceholder.typicode.com/posts/3",
]
for url in urls:
data = fetch_data(url)
print(f"Got {len(data)} items from {url}")
print(f"Total time: {time.time() - start:.2f}s")
main()
In this synchronous version, each request blocks the program until it completes. If each request takes 1 second, the total time is approximately 3 seconds — the program sits idle for most of that time, waiting for network I/O.
With asynchronous code using asyncio and aiohttp, we can run these requests concurrently:
import asyncio
import aiohttp
import time
async def fetch_data(session, url):
print(f"Fetching {url}...")
async with session.get(url) as response:
return await response.json()
async def main():
start = time.time()
urls = [
"https://jsonplaceholder.typicode.com/posts/1",
"https://jsonplaceholder.typicode.com/posts/2",
"https://jsonplaceholder.typicode.com/posts/3",
]
async with aiohttp.ClientSession() as session:
tasks = [fetch_data(session, url) for url in urls]
results = await asyncio.gather(*tasks)
for url, data in zip(urls, results):
print(f"Got {len(data)} items from {url}")
print(f"Total time: {time.time() - start:.2f}s")
asyncio.run(main())
The async version runs all three requests concurrently, completing in roughly the same time as the slowest single request — about 1 second instead of 3.
Core Concepts: Coroutines, Awaitables, and the Event Loop
To write asynchronous Python code, you need to understand three key concepts:
1. Coroutines
A coroutine is a function defined with async def. When called, it returns a coroutine object that must be awaited or scheduled for execution:
async def greet(name):
return f"Hello, {name}!"
# This returns a coroutine object, not the result
coro = greet("Alice")
print(coro) # <coroutine object greet at 0x...>
# To actually run it, you need to await it
result = await coro
print(result) # "Hello, Alice!"
2. Awaitables
An awaitable is any object that can be used with the await keyword. There are three types:
-
Coroutines (from
async def) - Tasks (wrapping coroutines for concurrent execution)
- Futures (low-level objects representing a pending result)
3. The Event Loop
The event loop is the engine that drives async Python. It manages a queue of tasks, running them cooperatively. When a task hits an await that performs I/O, it yields control back to the event loop, which can then run other tasks while waiting.
Python's asyncio.run() handles event loop creation and management for you in most cases.
Practical Patterns for Real-World Use
Pattern 1: Running Multiple Tasks Concurrently with asyncio.gather()
The most common pattern is running several tasks in parallel and collecting their results:
async def process_item(item):
# Simulate some async work
await asyncio.sleep(0.5)
return f"Processed: {item}"
async def main():
items = ["A", "B", "C", "D", "E"]
results = await asyncio.gather(
*[process_item(item) for item in items]
)
print(results)
# ['Processed: A', 'Processed: B', 'Processed: C', 'Processed: D', 'Processed: E']
asyncio.run(main())
gather() runs all tasks concurrently and returns results in the same order as the input coroutines. If any task raises an exception, it propagates immediately unless you set return_exceptions=True.
Pattern 2: Task Groups for Structured Concurrency
Python 3.11 introduced TaskGroup, which provides better error handling and resource cleanup:
async def main():
async with asyncio.TaskGroup() as tg:
task1 = tg.create_task(process_item("A"))
task2 = tg.create_task(process_item("B"))
task3 = tg.create_task(process_item("C"))
# All tasks complete before continuing
print(f"Results: {task1.result()}, {task2.result()}, {task3.result()}")
Task groups ensure that if any task fails, all sibling tasks are cancelled automatically — no orphaned tasks.
Pattern 3: Timeout Handling with asyncio.wait_for()
Network operations can hang indefinitely. Always add timeouts:
async def slow_operation():
await asyncio.sleep(10)
return "Done"
async def main():
try:
result = await asyncio.wait_for(slow_operation(), timeout=3.0)
print(result)
except asyncio.TimeoutError:
print("Operation timed out after 3 seconds")
Pattern 4: Producer-Consumer with Queues
For processing streams of data, use asyncio.Queue:
import random
async def producer(queue):
for i in range(10):
item = f"item-{i}"
await queue.put(item)
print(f"Produced {item}")
await asyncio.sleep(random.uniform(0.1, 0.5))
await queue.put(None) # Sentinel to signal completion
async def consumer(queue, name):
while True:
item = await queue.get()
if item is None:
queue.task_done()
break
print(f"Consumer {name} processing {item}")
await asyncio.sleep(random.uniform(0.2, 0.4))
queue.task_done()
async def main():
queue = asyncio.Queue()
await asyncio.gather(
producer(queue),
consumer(queue, "A"),
consumer(queue, "B"),
)
asyncio.run(main())
Real-World Example: Async Web Scraper
Let's put it all together with a practical web scraper that fetches multiple pages concurrently:
import asyncio
import aiohttp
from dataclasses import dataclass
@dataclass
class PageResult:
url: str
status: int
size: int
content: str
async def fetch_page(session, url, timeout=10):
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=timeout)) as response:
content = await response.text()
return PageResult(url, response.status, len(content), content[:200])
except Exception as e:
return PageResult(url, 0, 0, str(e))
async def scrape_sites(urls, concurrency=5):
semaphore = asyncio.Semaphore(concurrency)
async def bounded_fetch(session, url):
async with semaphore:
return await fetch_page(session, url)
async with aiohttp.ClientSession() as session:
tasks = [bounded_fetch(session, url) for url in urls]
results = await asyncio.gather(*tasks)
return results
async def main():
urls = [
"https://example.com",
"https://httpbin.org/get",
"https://httpbin.org/status/200",
"https://httpbin.org/delay/1",
"https://httpbin.org/delay/2",
]
results = await scrape_sites(urls, concurrency=3)
for r in results:
print(f"{r.url:50s} | Status: {r.status} | Size: {r.size:>6} bytes")
asyncio.run(main())
Key points in this example:
- Semaphore limits concurrent connections to avoid overwhelming the target server
- Timeout handling prevents hanging on slow responses
- Dataclass provides clean structured results
- Error resilience catches exceptions per-page without crashing the entire batch
Common Pitfalls and How to Avoid Them
Pitfall 1: Blocking the Event Loop
Calling a blocking function (like time.sleep(), requests.get(), or file I/O) inside an async function blocks the entire event loop:
# ❌ BAD: blocks the event loop
async def bad_example():
time.sleep(1) # Blocks everything!
return "done"
# ✅ GOOD: use async alternatives
async def good_example():
await asyncio.sleep(1) # Yields control to event loop
return "done"
For CPU-intensive work, use asyncio.to_thread() to run it in a separate thread:
import hashlib
def intensive_hash(data):
return hashlib.sha256(data * 1000000).hexdigest()
async def main():
result = await asyncio.to_thread(intensive_hash, b"test data")
print(result)
Pitfall 2: Using Synchronous Libraries
The requests library is synchronous and blocks the event loop. Always use async-compatible libraries:
| Task | Synchronous | Asynchronous |
|---|---|---|
| HTTP requests | requests |
aiohttp, httpx
|
| File I/O | open() |
aiofiles |
| Database |
psycopg2, pymongo
|
asyncpg, motor
|
| Web frameworks |
Flask, Django
|
FastAPI, aiohttp
|
Pitfall 3: Forgetting to Await
A common mistake is forgetting await:
async def main():
task = some_coroutine() # ❌ Creates coroutine but never runs it
# ...
await task # ✅ Must await to actually execute
When Should You Actually Use Async?
Async is not always the answer. Here's a quick decision guide:
Use async when:
- Making many network requests (APIs, web scraping, microservices)
- Building web servers handling many concurrent connections
- Working with async databases or message queues
- Processing streams of data from multiple sources
Stick with sync when:
- Your program is mostly CPU-bound (image processing, video encoding)
- You're writing simple scripts or CLI tools
- Your team isn't familiar with async patterns
- The overhead of the event loop outweighs concurrency benefits
Conclusion
Python's async/await gives you a powerful tool for writing concurrent, I/O-bound code that's still readable and maintainable. Start with asyncio.gather() for parallel tasks, always use async-compatible libraries, add timeouts to network operations, and never block the event loop.
The learning curve is real, but once you get comfortable with coroutines and the event loop, you'll wonder how you ever managed without them.
What async patterns do you use in your projects? Share your thoughts in the comments below.
Top comments (0)