AsyncIO Deep Dive

Reviewed & published by Brayan K

AsyncIO is the backbone of asynchronous programming in Python. To build high-performance systems — APIs, websocket servers, scrapers, automation pipelines, or distributed workers — you must fully understand how the event loop works, how Tasks provide concurrency, and how Futures act as low-level building blocks.

Part of the free Python course at LearnCodingFast — hands-on lessons with examples you run in your browser, plus practice exercises and a quick quiz.

What You'll Learn in This Lesson

• • How the asyncio event loop schedules and switches between coroutines
• • The difference between Tasks and Futures — and when to use each
• • How to run multiple async operations concurrently with asyncio.gather
• • How to create and cancel Tasks, and handle timeouts safely
• • How to use asyncio.Queue for producer-consumer pipelines
• • Common async patterns used in real APIs, scrapers, and websocket servers

🔥 1. What Exactly Is the Event Loop?

The event loop is a scheduler that repeatedly:

• Picks an awaitable that is ready to run
• Executes a small portion of it
• Pauses it when it awaits I/O
• Switches to the next ready task
• Handles callbacks, timers, and I/O events

It's the "orchestra conductor" of asynchronous execution.

import asyncio

async def main():
    print("Event loop running!")

asyncio.run(main())

# asyncio.run() does:
# 1. Create event loop
# 2. Run coroutine
# 3. Clean up loop
# 4. Close

# ✅ Expected output:
# Event loop running!

• Create event loop
• Run coroutine
• Clean up loop

⚙️ 2. Creating Coroutines (The Basics)

A coroutine is a function that can be paused:

import asyncio

async def fetch_user():
    await asyncio.sleep(1)
    return {"name": "Alice"}

async def main():
    coro = fetch_user()   # nothing has run yet — this is only a coroutine object
    print(type(coro))     # <class 'coroutine'>
    result = await coro   # NOW it runs, pausing for 1 second at the sleep
    print(result)         # {'name': 'Alice'}

asyncio.run(main())

# ✅ Expected output:
# <class 'coroutine'>
# {'name': 'Alice'}

Coroutines don't run until awaited or turned into a Task.

🧠 3. Tasks — The Core of Concurrency

A Task wraps a coroutine and schedules it on the event loop so it runs concurrently.

import asyncio

async def work():
    await asyncio.sleep(1)
    return "done"

async def main():
    task = asyncio.create_task(work())
    print("Task started...")
    result = await task
    print(result)

asyncio.run(main())

• create_task() schedules coroutine immediately
• Execution overlaps with the rest of the program
• Awaiting task retrieves result

This is how we achieve concurrency in a single thread.

⚡ 4. Running Multiple Tasks Concurrently

asyncio.gather() runs many tasks at once:

import asyncio

async def a(): await asyncio.sleep(1); return "A"
async def b(): await asyncio.sleep(1); return "B"

async def main():
    # await only works inside an async def — that is why this lives in main()
    results = await asyncio.gather(a(), b())
    print(results)      # ['A', 'B'] — order matches the order you passed them

asyncio.run(main())

Total runtime: 1 second, not 2.

• Fetching from many APIs
• Processing many files
• Running many workers
• Web scraping
• Database batch loading

🧪 Worked Example — Three Fetches, One Wait

Time to put coroutines, gather and the event loop together in one program you can actually run. Three fake "network calls" each wait a different amount of time. Run one after another they would take 0.3 + 0.1 + 0.2 = 0.6 seconds. Overlapped, they take about as long as the slowest one.

Read the comments first — each says what the line does and why it matters.

import asyncio
import time

# An "async def" function is a COROUTINE: a function allowed to pause in the
# middle and hand control back to the event loop while it is waiting.
async def fetch_user(user_id, delay):
    print(f"  start  user {user_id}")

    # asyncio.sleep stands in for real waiting — a network call, a database
    # query, a disk read. While this line waits, the event loop is free to run
    # the OTHER coroutines. That is the whole trick: waiting, not working.
    await asyncio.sleep(delay)

    print(f"  finish user {user_id}")
    return {"id": user_id, "name": f"User{user_id}"}


async def main():
    start = time.perf_counter()   # a high-resolution stopwatch

    # gather() schedules all three at once and waits for every one of them.
    # Results come back in the order you PASSED them, not the order they
    # finished — which is why user 1 is first in the list even though it
    # finished last.
    users = await asyncio.gather(
        fetch_user(1, 0.3),
        fetch_user(2, 0.1),
        fetch_user(3, 0.2),
    )

    elapsed = time.perf_counter() - start
    print("results:", users)
    print(f"elapsed: {elapsed:.1f}s")   # ~0.3s, not 0.6s


# asyncio.run() builds an event loop, runs main() on it, then closes it down.
asyncio.run(main())

# ✅ Expected output:
#   start  user 1
#   start  user 2
#   start  user 3
#   finish user 2
#   finish user 3
#   finish user 1
# results: [{'id': 1, 'name': 'User1'}, {'id': 2, 'name': 'User2'}, {'id': 3, 'name': 'User3'}]
# elapsed: 0.3s

Notice the three "start" lines print before any "finish" line. Every coroutine got going, hit its await, and stepped aside — that is concurrency in a single thread. The finish order follows the delays (0.1, 0.2, 0.3), while the results list keeps your original order.

🎯 Your Turn — Make Two Downloads Overlap

Everything is written except the three pieces this lesson is about: the keyword that turns a function into a coroutine, the keyword that pauses without blocking, and the function that runs both downloads at once. Fill in the blanks and check your output against the bottom of the file.

import asyncio

# 🎯 YOUR TURN — make these two downloads overlap instead of queueing up
# Fill in the three blanks marked ___

___ def download(name, seconds):     # 👉 the keyword that makes this a coroutine
    print(f"downloading {name}...")
    ___ asyncio.sleep(seconds)       # 👉 the keyword that pauses without blocking
    print(f"{name} done")
    return name.upper()


async def main():
    # 👉 replace ___ with the asyncio function that runs both at once
    files = await asyncio.___(download("notes.txt", 0.2), download("photo.png", 0.1))
    print(files)


asyncio.run(main())

# ✅ Expected output:
# downloading notes.txt...
# downloading photo.png...
# photo.png done
# notes.txt done
# ['NOTES.TXT', 'PHOTO.PNG']

If photo.png finishes after notes.txt, you have almost certainly awaited the two downloads one at a time instead of handing both to the same call.

🌀 5. Futures — Low-Level Awaitables

A Future represents a placeholder for a value that isn't available yet.

You rarely create Futures manually, but Tasks and event-loop internals rely on them.

import asyncio

async def main():
    loop = asyncio.get_running_loop()
    future = loop.create_future()

    loop.call_later(1, future.set_result, "Future complete")

    print(await future)

asyncio.run(main())

# ✅ Expected output:
# Future complete

This teaches two critical things:

• Futures hold results that arrive later
• Callbacks can resolve Futures

Tasks are built on Futures — every Task is a subclass of Future.

⏳ 6. Understanding How Tasks Progress

A task runs until it hits an await that yields control:

import asyncio

async def step1():
    print("Step 1")
    await asyncio.sleep(1)
    print("Step 1 done")

async def step2():
    print("Step 2")
    await asyncio.sleep(1)
    print("Step 2 done")

async def main():
    await asyncio.gather(step1(), step2())

asyncio.run(main())

• Step 1 runs → hits sleep → yields
• Step 2 runs → hits sleep → yields
• Event loop resumes Step 1 and Step 2

This overlapping execution is concurrency.

🧩 7. Task Cancellation

Every real system must handle cancellations:

import asyncio

async def worker():
    try:
        while True:
            await asyncio.sleep(1)
            print("Working...")
    except asyncio.CancelledError:
        print("Task cancelled!")

async def main():
    task = asyncio.create_task(worker())
    await asyncio.sleep(3)
    task.cancel()
    await task

asyncio.run(main())

• Shutting down servers
• Stopping background loops gracefully

🧱 8. Task Groups (Python 3.11+)

One of the newest and cleanest APIs:

import asyncio

async def fetch_data():
    await asyncio.sleep(0.2)
    print("data ready")

async def fetch_user():
    await asyncio.sleep(0.1)
    print("user ready")

async def main():
    # The block does not exit until every task inside it has finished.
    async with asyncio.TaskGroup() as tg:      # Python 3.11+
        tg.create_task(fetch_data())
        tg.create_task(fetch_user())
    print("both finished")

asyncio.run(main())

# ✅ Expected output:
# user ready
# data ready
# both finished

• Automatic error propagation
• Structured concurrency
• Cleaner code than gather()

⚡ 9. Wait vs Gather — When To Use Which?

• Returns results
• Cancels all tasks if one fails
• Best for symmetric jobs
• More control
• Choose FIRST_COMPLETED, FIRST_EXCEPTION
• Best for: Race conditions
• Redundant API fetches
• Timeout logic

done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)

⏱️ 10. Using Timeouts Correctly

try:
    await asyncio.wait_for(task, timeout=3)
except asyncio.TimeoutError:
    print("Timed out!")

Timeouts are essential for robust production systems.

🔄 11. Callbacks & Event Loop Scheduling

You can schedule code without async:

loop.call_later(2, lambda: print("Hello 2s later"))
loop.call_soon(lambda: print("Hello ASAP"))

This gives event-loop-level control that frameworks use internally.

🛰️ 12. Real-World Example — Concurrent API Fetching

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as s:
        async with s.get(url) as r:
            return await r.json()

async def main():
    urls = [
        "https://api1.com",
        "https://api2.com",
        "https://api3.com",
    ]
    results = await asyncio.gather(*(fetch(u) for u in urls))
    print(results)

asyncio.run(main())

This is how modern backend services fetch data from multiple microservices at once.

📡 13. Real-World Example — WebScraping With Concurrency

import asyncio
import aiohttp

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

async def scrape_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch(session, u)) for u in urls]
        return await asyncio.gather(*tasks)

This pattern lets you scrape hundreds of pages per second.

🔥 14. Production Architecture Using Tasks

A real backend service might have:

• Session cleanup
• Cache warmers
• Message queue consumers
• API request-response cycles
• Graceful shutdown
• Task cancellation
• Producer/consumer pipelines

All run on the same event loop.

🎯 Mini-Challenge: Give a Slow Job a Deadline

No code this time — just an outline. Write a job that takes 0.4 seconds, then call it twice through asyncio.wait_for: once with a deadline it cannot meet, once with a deadline it can. Catching asyncio.TimeoutError is the part that matters; a timeout you do not catch will crash the program.

# 🎯 MINI-CHALLENGE: give a slow job a deadline
#
# 1. import asyncio
# 2. async def slow_job():  wait 0.4 seconds, then return the string "finished"
# 3. async def main():
#      a) try to await asyncio.wait_for(slow_job(), timeout=0.1)
#         - print the result if it comes back
#         - except asyncio.TimeoutError:  print  timed out after 0.1s
#      b) do the same again with timeout=1.0, which this time will succeed
# 4. asyncio.run(main())
#
# ✅ Expected output:
# timed out after 0.1s
# finished

# your code here

Watch the indentation: except asyncio.TimeoutError: must line up with its own try:, and each attempt needs its own try block, otherwise the first timeout skips the second attempt entirely.

🎉 Conclusion

You've mastered three critical components of AsyncIO:

How async tasks are scheduled and run

Concurrent execution wrappers built on Futures

Low-level placeholders controlling async flow

Together, these form the foundation of every major async Python framework (FastAPI, Starlette, aiohttp).

📋 Quick Reference — AsyncIO

🎉 Great work! You've completed this lesson.

You now know how the asyncio event loop works internally, how to manage Tasks, and how to build async pipelines.

Practice quiz

Continue this course

• Previous: Advanced Async & Await Patterns
• Next: Concurrency in Python: Threads vs Processes — Understand the GIL and choose between threading and multiprocessing
• Quick reference: Python cheat sheet