asyncio / async / await

import asyncio

# Define an async function (coroutine)
async def my_coroutine():
    await asyncio.sleep(1) # Wait asynchronously
    return result

# Run a coroutine
asyncio.run(my_coroutine())

# Run multiple coroutines concurrently
results = await asyncio.gather(coro1(), coro2(), coro3())

asyncio.py

import asyncio
import time

# Basic async function
async def say_hello(name, delay):
    await asyncio.sleep(delay) # Other coroutines can run during this wait
    print(f"Hello, {name}!")

async def main():
    # Sequential execution (takes 3 seconds total)
    await say_hello('Kiryu Kazuma', 1)
    await say_hello('Majima Goro', 2)

asyncio.run(main())

# gather: concurrent execution (takes 2 seconds — the longest one)
async def main_parallel():
    start = time.time()
    await asyncio.gather(
        say_hello('Kiryu Kazuma', 1),
        say_hello('Majima Goro', 2),
        say_hello('Akiyama Shun', 1.5),
    )
    print(f"Total time: {time.time() - start:.1f}s") # About 2.0s

asyncio.run(main_parallel())

# create_task: schedule tasks to start immediately
async def fetch_data(url, delay):
    await asyncio.sleep(delay)
    return f"Data from {url}"

async def main_tasks():
    # Register as tasks (they start right away)
    task1 = asyncio.create_task(fetch_data('https://api1.example.com', 1))
    task2 = asyncio.create_task(fetch_data('https://api2.example.com', 2))

    result1 = await task1
    result2 = await task2
    print(result1)
    print(result2)

asyncio.run(main_tasks())

# asyncio.Queue: producer-consumer pattern
async def producer(q, items):
    for item in items:
        await q.put(item)
        await asyncio.sleep(0.5)
    await q.put(None) # Sentinel value to signal completion

async def consumer(q):
    while True:
        item = await q.get()
        if item is None:
            break
        print(f"Processing: {item}")
        q.task_done()

async def main_queue():
    q = asyncio.Queue()
    await asyncio.gather(
        producer(q, [1, 2, 3, 4, 5]),
        consumer(q),
    )

asyncio.run(main_queue())

Running the code produces the following output:

python3 asyncio.py
Hello, Kiryu Kazuma!
Hello, Majima Goro!
Hello, Kiryu Kazuma!
Hello, Akiyama Shun!
Hello, Majima Goro!
Total time: 2.0s
Data from https://api1.example.com
Data from https://api2.example.com
Processing: 1
Processing: 2
Processing: 3
Processing: 4
Processing: 5

fetch_parallel.py

import asyncio

# Pattern for concurrent requests using httpx (async HTTP client).
# Install: pip install httpx
async def fetch(client, url):
    try:
        response = await client.get(url, timeout=5.0)
        return {"url": url, "status": response.status_code}
    except Exception as e:
        return {"url": url, "error": str(e)}

async def fetch_all(urls):
    import httpx
    async with httpx.AsyncClient() as client:
        tasks = [fetch(client, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

urls = [
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/2",
    "https://httpbin.org/status/200",
]

# Use asyncio.run() to call the async entry point.
results = asyncio.run(fetch_all(urls))
for r in results:
    print(r)

Install the required package first:

pip install httpx
python3 fetch_parallel.py
{'url': 'https://httpbin.org/delay/1', 'status': 200}
{'url': 'https://httpbin.org/delay/2', 'status': 200}
{'url': 'https://httpbin.org/status/200', 'status': 200}

Sends concurrent requests to three URLs and waits for all of them to complete. Each result shows the URL and its HTTP status code. If a request fails, the error key contains the error message instead.

Common mistake 1: using time.sleep()

Using time.sleep() inside async def blocks the entire event loop. Always use asyncio.sleep() instead.

import asyncio

# NG: Using time.sleep() inside async def blocks the entire event loop.
async def bad_wait():
    import time
    time.sleep(2) # blocking! no other coroutine can run during this.

The corrected version is:

import asyncio

# OK: Always use asyncio.sleep() instead.
async def good_wait():
    await asyncio.sleep(2) # yields control to other coroutines.

Common mistake 2: missing await

Calling a coroutine without await does not execute it immediately — a RuntimeWarning is raised.

import asyncio

# NG: Calling a coroutine without await does not execute it immediately.
async def main():
    asyncio.sleep(1) # warning: coroutine never awaited (RuntimeWarning).

The corrected version is:

import asyncio

# OK: Always await coroutines.
async def main():
    await asyncio.sleep(1)

Common mistake 3: nested asyncio.run()

Calling asyncio.run() inside an already-running event loop raises a RuntimeError.

asyncio_run_ng.py

import asyncio

# NG: Calling asyncio.run() inside an already-running event loop raises an error.
async def outer():
    asyncio.run(inner()) # RuntimeError: This event loop is already running.

Running the code produces the following output:

python3 asyncio_run_ng.py
Traceback (most recent call last):
  ...
RuntimeError: This event loop is already running.

The corrected version is:

import asyncio

# OK: Use await inside async def.
async def outer():
    await inner()

Every function that uses asyncio must be a coroutine (async def). You cannot use await inside a regular function (def).

asyncio uses a single-threaded event loop that switches between multiple coroutines to run them concurrently. When execution reaches an await, the event loop hands control to another coroutine, allowing I/O wait time to be used efficiently.

asyncio.gather() runs all coroutines concurrently and returns their results as a list. If any coroutine raises an exception, that exception is propagated. To suppress exceptions instead, pass return_exceptions=True.

asyncio is powerful for I/O-bound tasks, but it does not help with CPU-bound work — heavy computation blocks the event loop. To combine asyncio with CPU-bound processing, use asyncio.to_thread() (Python 3.9+) to delegate work to a thread. asyncio is also commonly paired with async-capable HTTP client libraries such as httpx or aiohttp.