Tutorial: Learn Python Async with Asyncio from Scratch (2026)
The first time I refactored a synchronous web scraper to use asyncio, the runtime dropped from 45 seconds to 6 seconds — not by optimizing anything, just by letting I/O wait concurrently. Asyncio is Python's built-in library for writing concurrent code using the async/await syntax. It doesn't give you true parallelism (that's threading or multiprocessing), but it lets a single thread juggle many I/O-bound tasks efficiently by yielding control at await points.
This tutorial covers the asyncio mental model from the ground up: coroutines, event loops, tasks, and the main primitives like gather, create_task, and as_completed. We'll build an async web crawler using aiohttp that fetches multiple pages concurrently, handle timeouts and error recovery, and see why async is the right tool for network-bound work.
Coroutines and the Async/Await Syntax
A coroutine is a function defined with async def. Calling it returns a coroutine object, not the result. You execute it by awaiting it from another coroutine or by passing it to the event loop. The await keyword suspends the current coroutine until the awaited operation completes, allowing the event loop to run other tasks in the meantime. Any blocking call inside a coroutine blocks the entire event loop — never use time.sleep(), use asyncio.sleep().
import asyncio
async def greet(name):
await asyncio.sleep(1)
return f"Hello, {name}!"
async def main():
result = await greet("Alice")
print(result) # Hello, Alice! (after 1 second)
# Run
asyncio.run(main())The Event Loop and Running Coroutines
The event loop is the core scheduler that runs coroutines, handles I/O readiness notifications, and manages timers. asyncio.run() creates a new event loop, runs the main coroutine, and cleans up. For more control, use loop = asyncio.new_event_loop() and loop.run_until_complete(). The loop also handles subprocesses, signals, and file descriptor events through its selector.
import asyncio
async def slow_operation(n):
await asyncio.sleep(n)
print(f"Finished after {n}s")
async def main():
# Run multiple operations concurrently
await asyncio.gather(
slow_operation(3),
slow_operation(1),
slow_operation(2),
)
asyncio.run(main())
# Output:
# Finished after 1s
# Finished after 2s
# Finished after 3sTasks: Managing Concurrent Work
asyncio.create_task() schedules a coroutine to run on the event loop and returns a Task object. Tasks run concurrently — control returns to the caller immediately while the task executes in the background. Use await task to wait for completion and get the result, or task.cancel() to abort. Tasks are the building block for fan-out patterns where you kick off many operations and collect results as they complete.
async def fetch_url(url):
print(f"Fetching {url}")
await asyncio.sleep(1)
return f"Data from {url}"
async def main():
tasks = [
asyncio.create_task(fetch_url(f"https://site{i}.com"))
for i in range(5)
]
results = await asyncio.gather(*tasks)
for r in results:
print(r)
asyncio.run(main())Async HTTP with aiohttp
aiohttp is the standard async HTTP client for Python. It provides an asynchronous context manager for sessions, which reuses connection pools and supports keep-alive. Making requests is as simple as session.get(), session.post(), etc. I always use async with for both the session and the response to ensure proper resource cleanup. Timeouts are critical — use asyncio.wait_for() or aiohttp.ClientTimeout to prevent hanging.
import aiohttp
import asyncio
async def fetch_page(session, url):
async with session.get(url) as resp:
return await resp.text()
async def main():
timeout = aiohttp.ClientTimeout(total=10)
async with aiohttp.ClientSession(timeout=timeout) as session:
html = await fetch_page(session, "https://example.com")
print(len(html))
asyncio.run(main())
# Multiple concurrent fetches
async def fetch_all(urls):
async with aiohttp.ClientSession() as session:
tasks = [fetch_page(session, u) for u in urls]
return await asyncio.gather(*tasks)Error Handling and Timeouts
Exceptions in coroutines propagate normally with try/except, but concurrent tasks can fail independently. asyncio.gather(return_exceptions=True) returns exception objects instead of raising them. For timeouts, wrap any awaitable with asyncio.wait_for(coro, timeout). If the coroutine doesn't complete within the timeout, it raises asyncio.TimeoutError and cancels the task automatically.
import asyncio
async def unstable_operation(n):
await asyncio.sleep(n)
if n > 2:
raise ValueError(f"Too slow: {n}s")
return n
async def main():
results = await asyncio.gather(
unstable_operation(1),
unstable_operation(3),
unstable_operation(2),
return_exceptions=True
)
for r in results:
if isinstance(r, Exception):
print(f"Failed: {r}")
else:
print(f"Success: {r}")
asyncio.run(main())Async Context Managers and Async Iterators
Async context managers (async with) let you manage resources that require async setup and teardown — database connections, file handles, network streams. Define them with __aenter__ and __aexit__ coroutines. Async iterators (async for) let you consume async streams like websocket messages or paginated API responses. The @contextlib.asynccontextmanager decorator simplifies creating async context managers.
from contextlib import asynccontextmanager
@asynccontextmanager
async def managed_resource(name):
print(f"Acquiring {name}")
await asyncio.sleep(0.5)
yield {"name": name}
print(f"Releasing {name}")
async def main():
async with managed_resource("database") as res:
print(f"Using {res['name']}")
# Async iterator
async def counter():
for i in range(5):
await asyncio.sleep(0.2)
yield i
async def main2():
async for i in counter():
print(i)
asyncio.run(main())Frequently Asked Questions
Is asyncio faster than threading?
Asyncio has less overhead than threads (no GIL contention, no OS thread switching). For I/O-bound workloads with many concurrent connections, asyncio can handle tens of thousands of tasks where threading would be limited by OS thread count.
Can I mix sync and async code?
Yes, use loop.run_in_executor(None, sync_function) to run blocking code in a thread pool. Or use asyncio.to_thread(coro) in Python 3.9+. For calling async from sync, use asyncio.run(coro) but be careful about nested event loops.
What are common asyncio mistakes?
Blocking the event loop with time.sleep() or CPU-heavy code, forgetting to await a coroutine (returns a coroutine object), sharing mutable state between tasks without synchronization, and not setting timeouts on network calls.
Should I use asyncio or trio/anyio?
Asyncio is stdlib and the most widely supported in libraries. Trio offers a cleaner design with nurseries and structured concurrency. Anyio provides a compatibility layer. For new projects, consider trio for correctness; for ecosystem compatibility, stick with asyncio.
Originally published on Ayodhyyya. Last updated June 1, 2026.