Practice Now
Python
asyncio
programming
training
exercises

Python asyncio training exercises

Interview Questions practice on Codemia

Over 8,000 real interview questions from top companies, searchable by company and role.

Browse interview questions
Python

Introduction

Python's asyncio module provides an event loop for writing concurrent code using async/await syntax. It is designed for I/O-bound workloads — network requests, file operations, database queries — where the program spends most of its time waiting. Learning asyncio through hands-on exercises builds intuition for how coroutines schedule, how await suspends execution, and how to structure concurrent tasks. The exercises below progress from basic coroutines to real-world patterns like concurrent HTTP requests and producer-consumer queues.

Exercise 1: Basic Coroutine

python
1import asyncio
2
3async def greet(name, delay):
4    await asyncio.sleep(delay)
5    print(f"Hello, {name}!")
6
7async def main():
8    await greet("Alice", 1)
9    await greet("Bob", 2)
10    # Total time: ~3 seconds (sequential)
11
12asyncio.run(main())

This runs two coroutines sequentially. Each await suspends main() until the coroutine completes. The total time is the sum of both delays.

Exercise 2: Concurrent Tasks with gather

python
1import asyncio
2import time
3
4async def fetch(url, delay):
5    print(f"Fetching {url}...")
6    await asyncio.sleep(delay)
7    return f"Data from {url}"
8
9async def main():
10    start = time.time()
11
12    results = await asyncio.gather(
13        fetch("api/users", 2),
14        fetch("api/posts", 1),
15        fetch("api/comments", 3),
16    )
17
18    elapsed = time.time() - start
19    print(f"Results: {results}")
20    print(f"Time: {elapsed:.1f}s")  # ~3s, not 6s
21
22asyncio.run(main())

asyncio.gather() runs coroutines concurrently. The total time equals the longest task, not the sum. This is the most common pattern for parallelizing I/O-bound operations.

Exercise 3: Handling Exceptions in Tasks

python
1import asyncio
2
3async def risky_task(n):
4    await asyncio.sleep(1)
5    if n == 2:
6        raise ValueError(f"Task {n} failed")
7    return f"Task {n} done"
8
9async def main():
10    # gather with return_exceptions=True
11    results = await asyncio.gather(
12        risky_task(1),
13        risky_task(2),
14        risky_task(3),
15        return_exceptions=True,
16    )
17
18    for r in results:
19        if isinstance(r, Exception):
20            print(f"Error: {r}")
21        else:
22            print(r)
23
24asyncio.run(main())
25# Output:
26# Task 1 done
27# Error: Task 2 failed
28# Task 3 done

Without return_exceptions=True, gather raises the first exception and cancels remaining tasks. With it, exceptions are returned as results, letting you handle each task's outcome individually.

Exercise 4: Producer-Consumer with asyncio.Queue

python
1import asyncio
2import random
3
4async def producer(queue, name):
5    for i in range(5):
6        item = f"{name}-item-{i}"
7        await asyncio.sleep(random.uniform(0.1, 0.5))
8        await queue.put(item)
9        print(f"Produced: {item}")
10    await queue.put(None)  # Sentinel
11
12async def consumer(queue, name):
13    while True:
14        item = await queue.get()
15        if item is None:
16            break
17        await asyncio.sleep(random.uniform(0.2, 0.6))
18        print(f"{name} consumed: {item}")
19        queue.task_done()
20
21async def main():
22    queue = asyncio.Queue(maxsize=3)
23
24    producers = [asyncio.create_task(producer(queue, f"P{i}")) for i in range(2)]
25    consumers = [asyncio.create_task(consumer(queue, f"C{i}")) for i in range(2)]
26
27    await asyncio.gather(*producers)
28    # Send sentinel for each consumer
29    for _ in consumers:
30        await queue.put(None)
31    await asyncio.gather(*consumers)
32
33asyncio.run(main())

asyncio.Queue provides backpressure — put() blocks when the queue is full, and get() blocks when empty. This is the standard pattern for coordinating async producers and consumers.

Exercise 5: Timeout and Cancellation

python
1import asyncio
2
3async def slow_operation():
4    print("Starting slow operation...")
5    await asyncio.sleep(10)
6    return "Completed"
7
8async def main():
9    # Timeout after 2 seconds
10    try:
11        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
12        print(result)
13    except asyncio.TimeoutError:
14        print("Operation timed out!")
15
16    # Manual cancellation
17    task = asyncio.create_task(slow_operation())
18    await asyncio.sleep(1)
19    task.cancel()
20
21    try:
22        await task
23    except asyncio.CancelledError:
24        print("Task was cancelled")
25
26asyncio.run(main())

asyncio.wait_for() wraps a coroutine with a timeout. task.cancel() sends a CancelledError to the coroutine at its next await point. Both are essential for building resilient async applications.

Exercise 6: Semaphore for Rate Limiting

python
1import asyncio
2
3async def fetch_with_limit(sem, url):
4    async with sem:
5        print(f"Fetching {url}")
6        await asyncio.sleep(1)  # Simulate network request
7        return f"Data from {url}"
8
9async def main():
10    sem = asyncio.Semaphore(3)  # Max 3 concurrent requests
11    urls = [f"https://api.example.com/item/{i}" for i in range(10)]
12
13    tasks = [fetch_with_limit(sem, url) for url in urls]
14    results = await asyncio.gather(*tasks)
15    print(f"Fetched {len(results)} items")
16
17asyncio.run(main())

A semaphore limits how many coroutines run concurrently. This prevents overwhelming an API or database with too many simultaneous connections.

Exercise 7: Real HTTP Requests with aiohttp

python
1import asyncio
2import aiohttp
3
4async def fetch_url(session, url):
5    async with session.get(url) as response:
6        return await response.text()
7
8async def main():
9    urls = [
10        "https://httpbin.org/delay/1",
11        "https://httpbin.org/delay/2",
12        "https://httpbin.org/delay/1",
13    ]
14
15    async with aiohttp.ClientSession() as session:
16        tasks = [fetch_url(session, url) for url in urls]
17        results = await asyncio.gather(*tasks)
18
19    for url, result in zip(urls, results):
20        print(f"{url}: {len(result)} bytes")
21
22asyncio.run(main())

aiohttp is the standard library for async HTTP in Python. Share a single ClientSession across requests for connection pooling.

Common Pitfalls

  • Forgetting to await coroutines: Calling async_func() without await returns a coroutine object instead of executing it. You get a RuntimeWarning: coroutine was never awaited message. Always await or wrap in create_task().
  • Blocking the event loop: Calling time.sleep(), requests.get(), or any synchronous I/O inside a coroutine blocks the entire event loop. Use await asyncio.sleep(), aiohttp, or asyncio.to_thread() for blocking calls.
  • Using asyncio.run() inside a running loop: asyncio.run() creates a new event loop and fails if one is already running (common in Jupyter notebooks). Use await directly or nest_asyncio in notebooks.
  • Not handling CancelledError: When a task is cancelled, CancelledError is raised at the next await. If you catch Exception broadly, you may accidentally swallow cancellation. Catch CancelledError explicitly and re-raise if needed.
  • Creating too many concurrent tasks: Launching 10,000 tasks with gather() without a semaphore opens 10,000 connections simultaneously. Use asyncio.Semaphore or asyncio.TaskGroup (Python 3.11+) with limits.

Summary

  • Use async def and await to define and run coroutines
  • asyncio.gather() runs multiple coroutines concurrently — total time equals the slowest task
  • asyncio.Queue coordinates producer-consumer workflows with backpressure
  • asyncio.wait_for() adds timeouts; task.cancel() stops running tasks
  • asyncio.Semaphore limits concurrent execution for rate limiting
  • Never block the event loop — use async libraries or asyncio.to_thread() for synchronous code
Related reading
Free course
Beginner
7 lessons
2 hours
Tackling System Design Interview Problems

A short course that equips you with the skills to approach system design interviews methodically.

Start the free course
Track what you have practised

A free account saves your progress, solutions and study plan across every problem on Codemia.

Interview Questions practice on Codemia

Over 8,000 real interview questions from top companies, searchable by company and role.

Browse interview questions
Codemia
© 2026 Codemia
Resources
Legal
Social
All Rights Reserved.