Async Programming
Master Python asynchronous programming with async/await, asyncio, coroutines, and concurrent I/O for high-performance network applications.
Asynchronous programming lets Python handle many I/O-bound tasks concurrently without threads. It’s ideal for web servers, API clients, and any workload waiting on network or disk.
Sync vs Async
Synchronous code blocks until each operation completes:
import requests
def fetch_all(urls):
results = []
for url in urls:
results.append(requests.get(url).text) # blocks each time
return results
Asynchronous code can start the next request while waiting for the previous one:
import asyncio
import aiohttp
async def fetch(session, url):
async with session.get(url) as response:
return await response.text()
async def fetch_all(urls):
async with aiohttp.ClientSession() as session:
tasks = [fetch(session, url) for url in urls]
return await asyncio.gather(*tasks)
Coroutines with async/await
A coroutine is defined with async def and called with await:
import asyncio
async def say_hello():
print("Hello")
await asyncio.sleep(1) # non-blocking sleep
print("World")
asyncio.run(say_hello())
async def— defines a coroutine functionawait— suspends execution until the awaited coroutine completesasyncio.run()— entry point to run the top-level coroutine
Running Multiple Tasks
import asyncio
async def task(name, delay):
print(f"{name} starting")
await asyncio.sleep(delay)
print(f"{name} done")
return name
async def main():
results = await asyncio.gather(
task("A", 2),
task("B", 1),
task("C", 3),
)
print(results) # ['A', 'B', 'C']
asyncio.run(main())
asyncio.gather() runs coroutines concurrently and collects results.
Async Context Managers
Use async with for resources that support async cleanup:
import aiofiles
async def read_file(path):
async with aiofiles.open(path, 'r') as f:
return await f.read()
Async Iterators
async def async_range(n):
for i in range(n):
await asyncio.sleep(0.1)
yield i
async def main():
async for value in async_range(5):
print(value)
Building an Async Web Server
from aiohttp import web
async def handle(request):
name = request.match_info.get('name', 'World')
return web.Response(text=f"Hello, {name}!")
app = web.Application()
app.router.add_get('/{name}', handle)
# Run with: python -m aiohttp.web -P 8080 module:app
For production APIs, prefer FastAPI (built on asyncio and Starlette).
When to Use Async
| Use async when… | Use sync when… |
|---|---|
| Many concurrent I/O operations | CPU-bound computation |
| Web servers / API clients | Simple scripts |
| WebSocket connections | Libraries without async support |
| Database connection pools | Prototyping quickly |
Common Pitfalls
- Don’t call blocking code inside coroutines — use
asyncio.to_thread()or run in an executor. - Don’t mix sync and async carelessly — calling
asyncio.run()inside an already-running loop raises errors. - Async doesn’t speed up CPU work — use
multiprocessingfor compute-heavy tasks.
import asyncio
async def main():
# Run blocking function in a thread pool
result = await asyncio.to_thread(blocking_function, arg1, arg2)
Async Python unlocks high-concurrency I/O without the complexity of manual thread management.