Python · Syntax · Advanced

Async Python

10 tasks

Write concurrent programs using `async def`, `await`, and `asyncio`. Covers coroutines, `asyncio.gather`, tasks, `asyncio.Queue`, and async iterators.

Event loop, coroutines, await, and gather in depth

#
**Why async? The problem with blocking I/O** A typical server spends most of its time *waiting* — waiting for the database to respond, waiting for an HTTP request to arrive, waiting for a file to read. With normal synchronous code, the thread is blocked during that wait and can't do anything else. Async solves this by letting a single thread handle many waiting tasks at once. While one task is waiting for I/O, the event loop runs another task. No extra threads required. **What async is NOT** Async does not make CPU-bound work faster. If your code spends time computing (sorting, image processing, ML inference), async buys you nothing — the thread is busy, not waiting. For CPU work, use `multiprocessing`. Async shines only for I/O-bound work. **Coroutines: the building block** A coroutine is a function defined with `async def`. Calling it does NOT run the body — it creates a coroutine object. The event loop runs the body when scheduled: ```python import asyncio async def fetch(url: str) -> str: await asyncio.sleep(1) # yields control back to event loop return f'data from {url}' # Does nothing by itself: coro = fetch('https://example.com') # creates a coroutine object # Run it: result = asyncio.run(fetch('https://example.com')) # 'data from ...' ``` **`await` — suspending and resuming** `await expr` does two things: it suspends the current coroutine and yields control to the event loop. When the awaited thing completes, the event loop resumes this coroutine from where it paused. You can only `await` inside `async def`. Awaiting a non-awaitable raises `TypeError`. ```python async def slow_task(name: str, delay: float): print(f'{name}: starting') await asyncio.sleep(delay) # gives up control here print(f'{name}: done after {delay}s') async def main(): # Sequential — total 3 seconds: await slow_task('A', 1) await slow_task('B', 2) asyncio.run(main()) ``` **`asyncio.gather` — running tasks concurrently** `gather` schedules multiple coroutines to run concurrently in the same event loop. The event loop interleaves them while each waits for I/O: ```python async def main(): # Concurrent — total ~2 seconds (the longer one): results = await asyncio.gather( slow_task('A', 1), slow_task('B', 2), ) # A and B run concurrently, not one after the other asyncio.run(main()) ```

Tasks, async for/with, timeout, concurrent HTTP pattern

#
**`asyncio.Task` vs coroutine** A coroutine only runs when awaited. A `Task` wraps a coroutine and schedules it to run *independently* in the event loop — useful when you want to fire something and continue without immediately awaiting it: ```python import asyncio async def background_job(): await asyncio.sleep(5) print('background done') async def main(): task = asyncio.create_task(background_job()) # schedules immediately print('doing other work') # runs now, not after 5s await asyncio.sleep(1) print('still doing stuff') await task # wait for it to finish asyncio.run(main()) # doing other work # still doing stuff # background done (after 5s total) ``` **`async for` and `async with`** These work with objects that implement async protocols: ```python # async with — async context manager (__aenter__, __aexit__) async with aiofiles.open('data.txt') as f: content = await f.read() # async for — async iterator (__aiter__, __anext__) async for line in aiohttp_response.content: process(line) # async generator: async def lines(filename): async with aiofiles.open(filename) as f: async for line in f: yield line.strip() async def main(): async for line in lines('data.txt'): print(line) ``` **Timeout and cancellation** ```python import asyncio async def slow(): await asyncio.sleep(10) return 'result' async def main(): # Cancel if not done within 2 seconds: try: result = await asyncio.wait_for(slow(), timeout=2.0) except asyncio.TimeoutError: print('timed out') # Python 3.11+ asyncio.timeout: async with asyncio.timeout(2.0): result = await slow() # raises TimeoutError if >2s ``` **Real pattern: concurrent HTTP requests** ```python import asyncio import aiohttp async def fetch_url(session, url): async with session.get(url) as response: return await response.text() async def fetch_all(urls): async with aiohttp.ClientSession() as session: tasks = [fetch_url(session, url) for url in urls] return await asyncio.gather(*tasks) urls = ['https://example.com', 'https://example.org'] results = asyncio.run(fetch_all(urls)) # Both fetched concurrently — total time = slowest one, not sum ```

Common pitfalls, asyncio vs threads vs multiprocessing, quick ref

#
**Common async pitfalls** **1. Blocking the event loop** Any regular (non-async) blocking call freezes the entire event loop — all other tasks stop waiting while the call runs: ```python import asyncio, time async def bad(): time.sleep(5) # BLOCKS — freezes the whole event loop return 'done' async def good(): await asyncio.sleep(5) # yields to event loop return 'done' ``` For CPU work or unavoidable blocking calls, use `run_in_executor`: ```python import asyncio from concurrent.futures import ThreadPoolExecutor async def main(): loop = asyncio.get_event_loop() with ThreadPoolExecutor() as pool: result = await loop.run_in_executor(pool, blocking_io_call, arg) ``` **2. Forgetting to `await`** Calling a coroutine without `await` creates a coroutine object that never runs. Python 3.12+ and most linters warn about this: ```python async def fetch(): ... async def main(): result = fetch() # BUG: creates object, never runs result = await fetch() # correct ``` **3. Mixing sync and async code** You cannot `await` inside a regular function: ```python def sync_func(): await something() # SyntaxError: 'await' outside async function ``` If you need to call async code from sync context, use `asyncio.run()`. Don't call `asyncio.run()` from inside an already-running event loop. **asyncio vs threads vs multiprocessing — decision guide** | Workload | Solution | Why | |---|---|---| | I/O-bound (network, files, DB) | `asyncio` | Single thread, no overhead | | I/O-bound + blocking libs | `threading` | asyncio can't escape the GIL for blocking C code | | CPU-bound (compute-heavy) | `multiprocessing` | Bypasses the GIL; true parallelism | | Mixed I/O + CPU | `asyncio` + `run_in_executor` | Async for I/O, thread/process pool for CPU | **Quick reference** | Concept | Code | |---|---| | Define coroutine | `async def func(): ...` | | Await a coroutine | `result = await coro()` | | Run from sync | `asyncio.run(main())` | | Run concurrently | `asyncio.gather(c1, c2, c3)` | | Schedule independently | `asyncio.create_task(coro())` | | Timeout | `asyncio.wait_for(coro(), timeout=N)` | | Async context manager | `async with expr as x:` | | Async iterator | `async for item in expr:` |
01

First coroutine with asyncio.run

#

Write an `async def greet(name: str)` coroutine that prints `'Hello, <name>!'`, waits 0.1 seconds with `asyncio.sleep`, then prints `'Goodbye, <name>!'`. Run it with `asyncio.run(greet('Alice'))`.

import asyncio

async def greet(name: str) -> None:
    pass


asyncio.run(greet('Alice'))
# Hello, Alice!
# Goodbye, Alice!
Solution
import asyncio

async def greet(name: str) -> None:
    print(f'Hello, {name}!')
    await asyncio.sleep(0.1)
    print(f'Goodbye, {name}!')


asyncio.run(greet('Alice'))
# Hello, Alice!
# Goodbye, Alice!
02

Concurrent tasks with asyncio.gather

#

Write a coroutine `fetch(url: str, delay: float) -> str` that simulates an HTTP request by sleeping `delay` seconds and returning `f'Response from {url}'`. Use `asyncio.gather` to run three fetches concurrently: `fetch('A', 0.3)`, `fetch('B', 0.1)`, `fetch('C', 0.2)`. Print all results. The total time should be about 0.3s, not 0.6s.

import asyncio

async def fetch(url: str, delay: float) -> str:
    pass

async def main() -> None:
    results = await asyncio.gather(
        fetch('A', 0.3),
        fetch('B', 0.1),
        fetch('C', 0.2),
    )
    for r in results:
        print(r)


asyncio.run(main())
Solution
import asyncio

async def fetch(url: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f'Response from {url}'

async def main() -> None:
    results = await asyncio.gather(
        fetch('A', 0.3),
        fetch('B', 0.1),
        fetch('C', 0.2),
    )
    for r in results:
        print(r)


asyncio.run(main())
# Response from A
# Response from B
# Response from C
03

Background tasks with create_task

#

Write a `main()` coroutine that creates two background tasks using `asyncio.create_task`: one that counts down from 3 to 1 (printing each number with 0.1s delay), and one that prints `'Working...'` three times with 0.15s delay. Await both tasks at the end. Observe that the outputs interleave.

import asyncio

async def countdown() -> None:
    for i in range(3, 0, -1):
        print(f'Countdown: {i}')
        await asyncio.sleep(0.1)

async def worker() -> None:
    for _ in range(3):
        print('Working...')
        await asyncio.sleep(0.15)

async def main() -> None:
    # create tasks and await them
    pass


asyncio.run(main())
Solution
import asyncio

async def countdown() -> None:
    for i in range(3, 0, -1):
        print(f'Countdown: {i}')
        await asyncio.sleep(0.1)

async def worker() -> None:
    for _ in range(3):
        print('Working...')
        await asyncio.sleep(0.15)

async def main() -> None:
    task1 = asyncio.create_task(countdown())
    task2 = asyncio.create_task(worker())
    await task1
    await task2


asyncio.run(main())
04

Producer-consumer with asyncio.Queue

#

Implement a producer-consumer pattern using `asyncio.Queue`. The `producer` coroutine puts numbers 1–5 into the queue (with 0.05s delay each), then puts `None` as a sentinel. The `consumer` coroutine reads items from the queue until it sees `None`, printing `'Consuming: <n>'` for each number.

import asyncio

async def producer(queue: asyncio.Queue) -> None:
    pass

async def consumer(queue: asyncio.Queue) -> None:
    pass

async def main() -> None:
    queue: asyncio.Queue = asyncio.Queue()
    await asyncio.gather(producer(queue), consumer(queue))


asyncio.run(main())
# Consuming: 1
# Consuming: 2
# ...
Solution
import asyncio

async def producer(queue: asyncio.Queue) -> None:
    for i in range(1, 6):
        await queue.put(i)
        await asyncio.sleep(0.05)
    await queue.put(None)  # sentinel

async def consumer(queue: asyncio.Queue) -> None:
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f'Consuming: {item}')

async def main() -> None:
    queue: asyncio.Queue = asyncio.Queue()
    await asyncio.gather(producer(queue), consumer(queue))


asyncio.run(main())
05

Timeout with asyncio.wait_for

#

Write a coroutine `slow_op(seconds: float) -> str` that sleeps for `seconds` and returns `'done'`. Use `asyncio.wait_for(slow_op(5), timeout=0.2)` to attempt the operation with a 0.2-second timeout. Catch `asyncio.TimeoutError` and print `'Operation timed out!'`.

import asyncio

async def slow_op(seconds: float) -> str:
    await asyncio.sleep(seconds)
    return 'done'

async def main() -> None:
    try:
        result = await asyncio.wait_for(slow_op(5), timeout=0.2)
        print(result)
    except asyncio.TimeoutError:
        pass  # print the timeout message here


asyncio.run(main())
# Operation timed out!
Solution
import asyncio

async def slow_op(seconds: float) -> str:
    await asyncio.sleep(seconds)
    return 'done'

async def main() -> None:
    try:
        result = await asyncio.wait_for(slow_op(5), timeout=0.2)
        print(result)
    except asyncio.TimeoutError:
        print('Operation timed out!')


asyncio.run(main())
# Operation timed out!
06

Async generator

#

Write an async generator `async_range(start: int, stop: int, delay: float)` that yields integers from `start` to `stop-1`, pausing `delay` seconds between each yield. Use `async for` to consume it and print each value.

import asyncio

async def async_range(start: int, stop: int, delay: float):
    pass

async def main() -> None:
    async for value in async_range(0, 4, 0.05):
        print(value)


asyncio.run(main())
# 0
# 1
# 2
# 3
Solution
import asyncio

async def async_range(start: int, stop: int, delay: float):
    for i in range(start, stop):
        yield i
        await asyncio.sleep(delay)

async def main() -> None:
    async for value in async_range(0, 4, 0.05):
        print(value)


asyncio.run(main())
# 0
# 1
# 2
# 3
07

Async context manager

#

Create a class `AsyncTimer` that works as an async context manager. `__aenter__` records the start time and prints `'Timer started'`. `__aexit__` computes elapsed time and prints `f'Elapsed: {elapsed:.2f}s'`. Use it with `async with AsyncTimer():`.

import asyncio
import time

class AsyncTimer:
    async def __aenter__(self):
        pass

    async def __aexit__(self, *args):
        pass

async def main() -> None:
    async with AsyncTimer():
        await asyncio.sleep(0.1)


asyncio.run(main())
# Timer started
# Elapsed: 0.10s
Solution
import asyncio
import time

class AsyncTimer:
    async def __aenter__(self):
        self._start = time.perf_counter()
        print('Timer started')
        return self

    async def __aexit__(self, *args):
        elapsed = time.perf_counter() - self._start
        print(f'Elapsed: {elapsed:.2f}s')

async def main() -> None:
    async with AsyncTimer():
        await asyncio.sleep(0.1)


asyncio.run(main())
# Timer started
# Elapsed: 0.10s
08

Limit concurrency with asyncio.Semaphore

#

You have 8 tasks but want at most 3 running concurrently. Use `asyncio.Semaphore(3)` to limit concurrency. Each task should print `'Task N started'`, sleep 0.1s, then print `'Task N done'`. Create all 8 tasks with `asyncio.gather`.

import asyncio

async def task(n: int, sem: asyncio.Semaphore) -> None:
    async with sem:
        print(f'Task {n} started')
        await asyncio.sleep(0.1)
        print(f'Task {n} done')

async def main() -> None:
    sem = asyncio.Semaphore(3)
    # create and gather 8 tasks (n from 1 to 8)
    pass


asyncio.run(main())
Solution
import asyncio

async def task(n: int, sem: asyncio.Semaphore) -> None:
    async with sem:
        print(f'Task {n} started')
        await asyncio.sleep(0.1)
        print(f'Task {n} done')

async def main() -> None:
    sem = asyncio.Semaphore(3)
    await asyncio.gather(*(task(n, sem) for n in range(1, 9)))


asyncio.run(main())
09

Run blocking code with run_in_executor

#

The function `blocking_sum(n)` uses a CPU-bound loop to compute the sum of 1..n (simulating slow blocking work). Wrap it in an async function `async_sum(n)` using `loop.run_in_executor(None, blocking_sum, n)` so it doesn't block the event loop. Run two `async_sum` calls concurrently with `asyncio.gather`.

import asyncio

def blocking_sum(n: int) -> int:
    return sum(range(n + 1))

async def async_sum(n: int) -> int:
    loop = asyncio.get_running_loop()
    # run blocking_sum in a thread pool
    pass

async def main() -> None:
    results = await asyncio.gather(async_sum(1000), async_sum(2000))
    print(results)  # [500500, 2001000]


asyncio.run(main())
Solution
import asyncio

def blocking_sum(n: int) -> int:
    return sum(range(n + 1))

async def async_sum(n: int) -> int:
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, blocking_sum, n)

async def main() -> None:
    results = await asyncio.gather(async_sum(1000), async_sum(2000))
    print(results)  # [500500, 2001000]


asyncio.run(main())
10

Coordination with asyncio.Event

#

Use `asyncio.Event` to coordinate two coroutines. The `setter` coroutine waits 0.15s, prints `'Setting event'`, then sets the event. The `waiter` coroutine prints `'Waiting for event...'`, waits with `await event.wait()`, then prints `'Event received!'`. Run both concurrently with `asyncio.gather`.

import asyncio

async def setter(event: asyncio.Event) -> None:
    await asyncio.sleep(0.15)
    print('Setting event')
    event.set()

async def waiter(event: asyncio.Event) -> None:
    pass  # wait for the event, then print 'Event received!'

async def main() -> None:
    event = asyncio.Event()
    await asyncio.gather(setter(event), waiter(event))


asyncio.run(main())
# Waiting for event...
# Setting event
# Event received!
Solution
import asyncio

async def setter(event: asyncio.Event) -> None:
    await asyncio.sleep(0.15)
    print('Setting event')
    event.set()

async def waiter(event: asyncio.Event) -> None:
    print('Waiting for event...')
    await event.wait()
    print('Event received!')

async def main() -> None:
    event = asyncio.Event()
    await asyncio.gather(setter(event), waiter(event))


asyncio.run(main())
# Waiting for event...
# Setting event
# Event received!