Many programs spend much of their time waiting: for an HTTP response, a database query, a timer, or a file-like network stream. During that wait, a synchronous program may sit idle. Python asyncio provides a structured way to run many waiting operations cooperatively on a single thread, which can make I/O-bound applications responsive and efficient.
This practical guide explains coroutines, the event loop, tasks, concurrent waiting, timeouts, cancellation, error handling, and when asynchronous code is the wrong tool. It assumes you already understand regular Python functions.
What asynchronous programming solves
Asynchronous programming is useful when tasks frequently wait for external events. Instead of blocking the whole program, a coroutine can pause at an await expression and allow another coroutine to run. This is concurrency, not necessarily parallel execution.
The official asyncio documentation describes the library as a foundation for concurrent code using async/await. It is widely used for network services, clients, queues, subprocess control, and other I/O-heavy systems.
Your first coroutine
Define a coroutine function with async def. Calling it creates a coroutine object; it does not immediately execute the body. asyncio.run() creates an event loop, runs the top-level coroutine, and closes the loop.
import asyncio
async def greet() -> None:
print("Hello")
await asyncio.sleep(1)
print("Goodbye")
asyncio.run(greet())Unlike time.sleep(), asyncio.sleep() suspends only the current coroutine and gives the event loop a chance to run other work. The Python time module guide explains the blocking alternative.
Run independent operations concurrently
Awaiting two coroutines one after another is still sequential. To overlap independent waits, use asyncio.gather() or create tasks.
import asyncio
async def fetch(name: str, delay: float) -> str:
print(f"Starting {name}")
await asyncio.sleep(delay)
return f"{name} finished"
async def main() -> None:
results = await asyncio.gather(
fetch("users", 1.0),
fetch("orders", 1.5),
fetch("products", 0.5),
)
print(results)
asyncio.run(main())The total wait is close to the longest individual delay rather than the sum, because the operations are waiting concurrently. Real applications replace the simulated sleeps with asynchronous network or database libraries.
Create and manage tasks
asyncio.create_task() schedules a coroutine to run soon and returns a task object. Keep a reference and await it so exceptions are observed and the operation is allowed to finish.
import asyncio
async def heartbeat() -> None:
for number in range(3):
await asyncio.sleep(0.5)
print(f"heartbeat {number}")
async def main() -> None:
task = asyncio.create_task(heartbeat())
print("Main coroutine is doing other work")
await asyncio.sleep(0.75)
await task
asyncio.run(main())Creating “fire-and-forget” tasks without ownership often causes lost exceptions, premature shutdown, and difficult tests. A task should normally belong to a scope that waits for it, cancels it, or records its outcome.
Structured concurrency with TaskGroup
Modern Python offers asyncio.TaskGroup for a group of related tasks. The context manager waits for every task, and if one fails it cancels the remaining tasks and reports grouped errors. This makes task lifetime easier to reason about.
import asyncio
async def worker(name: str, delay: float) -> None:
await asyncio.sleep(delay)
print(f"{name} done")
async def main() -> None:
async with asyncio.TaskGroup() as group:
group.create_task(worker("A", 0.4))
group.create_task(worker("B", 0.2))
group.create_task(worker("C", 0.6))
asyncio.run(main())Timeouts
External systems can be slow or unavailable. A timeout prevents one operation from waiting forever. Use asyncio.timeout() to define a deadline around one or more awaited operations.
import asyncio
async def slow_operation() -> str:
await asyncio.sleep(5)
return "finished"
async def main() -> None:
try:
async with asyncio.timeout(1):
result = await slow_operation()
print(result)
except TimeoutError:
print("The operation exceeded its deadline")
asyncio.run(main())Choose timeouts according to real service expectations and make them configurable. A timeout is not a complete retry policy; retries need limits, delays, and careful treatment of operations that may have partially succeeded.
Cancellation and cleanup
Cancellation is a normal control-flow event in asynchronous programs. When a task is cancelled, an asyncio.CancelledError is raised inside it. Use finally to release resources, and generally re-raise cancellation instead of silently consuming it.
import asyncio
async def monitor() -> None:
try:
while True:
print("checking")
await asyncio.sleep(1)
finally:
print("monitor cleanup")
async def main() -> None:
task = asyncio.create_task(monitor())
await asyncio.sleep(2.2)
task.cancel()
try:
await task
except asyncio.CancelledError:
print("monitor cancelled")
asyncio.run(main())The same principle applies to files, sockets, and database connections: use context managers and explicit cleanup. See using with in Python for the synchronous foundation.
Protect shared state with a lock
Coroutines run cooperatively, but race conditions are still possible when an operation reads state, awaits, and then writes based on the old value. asyncio.Lock protects critical sections among coroutines in one event loop.
import asyncio
balance = 0
lock = asyncio.Lock()
async def add(amount: int) -> None:
global balance
async with lock:
current = balance
await asyncio.sleep(0)
balance = current + amount
async def main() -> None:
await asyncio.gather(*(add(1) for _ in range(100)))
print(balance)
asyncio.run(main())Minimize shared mutable state when possible. Passing values through queues or returning results can produce simpler designs than protecting many global variables.
Use an asynchronous queue
asyncio.Queue connects producers and consumers and provides backpressure through a maximum size. It is useful for crawlers, job processors, event pipelines, and batch services.
import asyncio
async def producer(queue: asyncio.Queue[int]) -> None:
for item in range(5):
await queue.put(item)
await queue.put(-1)
async def consumer(queue: asyncio.Queue[int]) -> None:
while True:
item = await queue.get()
try:
if item == -1:
return
print(f"processed {item}")
finally:
queue.task_done()
async def main() -> None:
queue: asyncio.Queue[int] = asyncio.Queue(maxsize=2)
await asyncio.gather(
producer(queue),
consumer(queue),
)
asyncio.run(main())Do not block the event loop
Regular blocking calls stop every coroutine on the event-loop thread. CPU-heavy loops, time.sleep(), synchronous HTTP clients, and slow filesystem operations can make an async application appear frozen. Use an asynchronous library or move unavoidable blocking work to a thread with asyncio.to_thread().
import asyncio
from pathlib import Path
def read_large_file() -> str:
return Path("report.txt").read_text(encoding="utf-8")
async def main() -> None:
text = await asyncio.to_thread(read_large_file)
print(len(text))
asyncio.run(main())Threads are not a shortcut for CPU-bound parallelism. For that distinction, see the guide explaining Python threading and the article on the Global Interpreter Lock.
Error handling
Catch exceptions at a level that can respond meaningfully. A low-level coroutine may add context and re-raise; a task group or request handler may decide whether to retry, return an error, or shut down. Avoid broad exception handlers that turn failures into silent missing results.
import asyncio
async def parse_remote_value(value: str) -> int:
await asyncio.sleep(0.1)
try:
return int(value)
except ValueError as error:
raise ValueError(f"Invalid remote integer: {value!r}") from error
async def main() -> None:
try:
result = await parse_remote_value("not-a-number")
except ValueError as error:
print(error)
else:
print(result)
asyncio.run(main())The try/except tutorial covers exception chaining and specific handlers in more depth.
When asyncio is a good fit
- Many network requests or long-lived connections.
- High-concurrency servers and clients.
- Pipelines with queues, timers, and asynchronous subprocesses.
- Applications already using async-compatible frameworks and drivers.
- Workloads dominated by waiting rather than computation.
When asyncio is not a good fit
- A small sequential script where async code would add ceremony.
- CPU-heavy numerical work that needs multiprocessing or native vectorization.
- Libraries that only expose blocking APIs and cannot be isolated cleanly.
- Codebases whose team cannot yet test and debug asynchronous behavior safely.
Frequently asked questions
Is asyncio multithreading?
Not by default. Most asyncio code runs on one event-loop thread and switches at await points. It can cooperate with threads through tools such as to_thread(), but the core model is different.
Can I call asyncio.run() inside a running event loop?
No. Interactive environments and async frameworks may already own the loop. In that context, await the coroutine directly or use the framework’s entry point.
Does async make every program faster?
No. It improves utilization when work spends time waiting and many operations can overlap. For a single quick operation or CPU-bound algorithm, asynchronous structure may add overhead without benefit.
Final thoughts
Python asyncio is easiest to understand as cooperative task management for I/O-bound work. Start with a single top-level asyncio.run(), await every operation you own, prefer structured task groups, set timeouts, preserve cancellation, and keep blocking code away from the event loop. Those habits make asynchronous systems far more predictable.
