Introduction

You have probably seen code like this:

async def fetch_data():
    await some_api_call()

The async and await keywords are part of Python’s asyncio framework — a way to write concurrent code that handles thousands of operations simultaneously without using threads.

But what does it actually mean? Why does it exist? And when should you use it instead of threading?

This article explains how asyncio works from first principles, then shows you how to use it effectively.

All examples are tested on Python 3.12.

---

The Problem asyncio Solves

Consider fetching data from 100 URLs one at a time:

import urllib.request
import time

urls = [f"https://httpbin.org/delay/1" for _ in range(5)]

start = time.time()
for url in urls:
    urllib.request.urlopen(url)
print(f"Sequential: {time.time() - start:.1f}s")

Example output:

Sequential: 5.1s

Each request waits for the previous one to finish. The program spends most of its time doing nothing — just waiting for the network.

asyncio solves this by letting Python do other work while waiting.

---

How asyncio Works: The Event Loop

asyncio runs on an event loop — a single thread that manages many operations simultaneously by switching between them whenever one is waiting.

The mental model:

Event loop checks:
  → Task A is waiting for network? Move on.
  → Task B is waiting for disk? Move on.
  → Task C has data ready! Run it.
  → Task A has data now! Run it.
  → ...

This is called cooperative multitasking. Tasks voluntarily yield control when they are waiting, using the await keyword. The event loop then runs another task.

---

Coroutines and async/await

A coroutine is a function defined with async def. It can be paused and resumed:

import asyncio

async def greet(name, delay):
    print(f"Hello, {name}!")
    await asyncio.sleep(delay)  # yields control here
    print(f"Goodbye, {name}!")

asyncio.run(greet("Alice", 1))

Expected output:

Hello, Alice!
Goodbye, Alice!

await asyncio.sleep(1) pauses the coroutine for 1 second. During that pause, the event loop can run other coroutines.

asyncio.run() creates an event loop, runs the coroutine, and closes the loop.

---

Running Coroutines Concurrently

The real power comes from running multiple coroutines at the same time:

import asyncio
import time

async def task(name, delay):
    print(f"{name} started")
    await asyncio.sleep(delay)
    print(f"{name} finished after {delay}s")

async def main():
    start = time.time()

    await asyncio.gather(
        task("A", 1),
        task("B", 2),
        task("C", 1),
    )

    print(f"Total: {time.time() - start:.1f}s")

asyncio.run(main())

Expected output:

A started
B started
C started
A finished after 1s
C finished after 1s
B finished after 2s
Total: 2.0s

All three tasks run concurrently. The total time is 2 seconds (the longest task), not 4 seconds (1+2+1).

asyncio.gather() runs multiple coroutines concurrently and waits for all of them to finish.

---

Tasks

A Task wraps a coroutine and schedules it to run on the event loop immediately:

import asyncio

async def slow_operation(name):
    await asyncio.sleep(1)
    return f"{name} done"

async def main():
    # Create tasks — they start immediately
    task1 = asyncio.create_task(slow_operation("Task 1"))
    task2 = asyncio.create_task(slow_operation("Task 2"))

    # Do other work here while tasks run...
    print("Tasks are running in background")

    # Wait for results
    result1 = await task1
    result2 = await task2

    print(result1)
    print(result2)

asyncio.run(main())

Expected output:

Tasks are running in background
Task 1 done
Task 2 done

---

Real Example: Concurrent HTTP Requests

import asyncio
import aiohttp
import time

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/1",
    ]

    start = time.time()

    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(*[fetch(session, url) for url in urls])

    print(f"Fetched {len(results)} URLs in {time.time() - start:.1f}s")

asyncio.run(main())

Install aiohttp first:

pip install aiohttp

Example output:

Fetched 5 URLs in 1.1s

Five requests that would take 5 seconds sequentially complete in just over 1 second.

---

async for and async with

Some objects require async for and async with instead of the regular versions:

import asyncio
import aiofiles  # pip install aiofiles

async def read_file():
    async with aiofiles.open("data.txt", "r") as f:
        async for line in f:
            print(line.strip())

asyncio.run(read_file())

Use these when the operation itself is asynchronous (network I/O, async file access). Regular for and with still work for synchronous operations inside async functions.

---

Common Patterns

Timeout

import asyncio

async def slow():
    await asyncio.sleep(10)
    return "done"

async def main():
    try:
        result = await asyncio.wait_for(slow(), timeout=2.0)
    except asyncio.TimeoutError:
        print("Operation timed out")

asyncio.run(main())

Expected output:

Operation timed out

Limiting Concurrency with Semaphore

When fetching hundreds of URLs, you do not want to send all requests at once. Use a semaphore to limit concurrency:

import asyncio
import aiohttp

async def fetch_with_limit(session, url, semaphore):
    async with semaphore:
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = [f"https://httpbin.org/get?n={i}" for i in range(20)]
    semaphore = asyncio.Semaphore(5)  # max 5 concurrent requests

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_limit(session, url, semaphore) for url in urls]
        results = await asyncio.gather(*tasks)

    print(f"Fetched {len(results)} URLs")

asyncio.run(main())

---

asyncio vs threading vs multiprocessing

This is the most important question when choosing a concurrency model:

Situation	Best Choice
I/O-bound: network, files, databases	asyncio or threading
CPU-bound: computation, image processing	multiprocessing
Many concurrent connections (1000+)	asyncio
Simple I/O with existing sync libraries	threading
True parallel CPU work	multiprocessing

asyncio wins for high-concurrency I/O where you control the code. It uses a single thread, so there is no thread overhead or locking.

threading is easier to retrofit onto existing synchronous code. The GIL limits CPU parallelism, but threading works well for I/O-bound tasks because the GIL is released during I/O waits.

multiprocessing bypasses the GIL entirely and is the right choice for CPU-heavy work.

---

When NOT to Use asyncio

asyncio is not always the answer:

When your libraries are not async-aware. Libraries like requests, psycopg2, or standard sqlite3 are synchronous. Calling them inside a coroutine blocks the entire event loop. Use their async equivalents (aiohttp, asyncpg, aiosqlite) instead.

When you have CPU-bound work. asyncio does not help with CPU-heavy computation — use multiprocessing.

When simplicity matters more than performance. For a script that fetches 10 URLs, threading is simpler and the performance difference is negligible.

---

Wrap-Up

asyncio works by running a single-threaded event loop that switches between coroutines whenever one is waiting. The async/await syntax marks where those switches can happen.

Three things to remember:

1. Use asyncio for I/O-bound code with high concurrency requirements
2. await yields control to the event loop — only use it inside async def functions
3. Use async-compatible libraries (aiohttp, aiofiles) — synchronous libraries block the event loop

For more on Python’s concurrency model and why the GIL matters, see the GIL deep dive. For questions or future tutorial ideas, get in touch via the Contact page.