Skip to main content

7.8 - Coroutines

7.8.1 - Introduction to Coroutines​

Coroutines are a powerful feature in Python that allows for efficient asynchronous programming. They enable functions to be paused and resumed, making them ideal for tasks that involve waiting for I/O operations. This guide will provide an understanding of coroutines, their syntax, and practical use cases.

7.8.2 - Understanding async and await​

7.8.2.1 - The async Keyword​

  • Definition: The async keyword is used to define a coroutine. It modifies a function, indicating that the function is an asynchronous coroutine rather than a normal function.
  • Usage: Placed before def, as in async def function_name():. This creates a coroutine function.
  • Behavior: When called, an async function doesn't execute immediately. Instead, it returns a coroutine object that can be executed asynchronously.

7.8.2.2 - The await Keyword​

  • Definition: await is used to pause the coroutine's execution until the awaited task is complete. It can only be used inside an async function.
  • Usage: Applied before calling a coroutine or any awaitable object, like await coroutine_name().
  • Behavior: Pauses the execution of the coroutine where it is used and waits for the awaited task to complete. During this pause, other tasks can run.

7.8.2.3 - Interaction and Workflow​

async and await work together to manage asynchronous tasks. When await is called in a coroutine, it tells the Python event loop to run other tasks until the awaited task completes, at which point the coroutine resumes.

7.8.3 - Creating and Running Coroutines​

To create a coroutine, use the async def statement. To run a coroutine, you must await it within another coroutine or use an event loop, like asyncio.run().

import asyncio

async def my_coroutine():
    print("My Coroutine")

# Running the coroutine with an event loop
asyncio.run(my_coroutine())

7.8.4 - Coroutines Case Studies​

7.8.4.1 - Asynchronous Web Request​

Handling web requests asynchronously is a common use case for coroutines. This example uses aiohttp to make non-blocking HTTP requests.

import aiohttp
import asyncio

async def fetch_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    html = await fetch_url("http://example.com")
    print(html)

asyncio.run(main())

7.8.4.2 - Asynchronous File Operations​

Performing file operations asynchronously can improve the efficiency of I/O-bound tasks.

async def read_file_async(file_path):
    await asyncio.sleep(1)  # Simulating an I/O-bound task
    with open(file_path, 'r') as file:
        return file.read()

async def main():
    content = await read_file_async('example.txt')
    print(content)

asyncio.run(main())

7.8.4.3 - Producer-Consumer Using Asyncio​

This demonstrates the producer-consumer pattern in asynchronous programming.

async def producer(queue):
    for i in range(10):
        await queue.put(i)
        print(f'Produced {i}')
        await asyncio.sleep(1)

async def consumer(queue):
    while True:
        item = await queue.get()
        print(f'Consumed {item}')
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    producer_coroutine = producer(queue)
    consumer_coroutine = consumer(queue)
    await asyncio.gather(producer_coroutine, consumer_coroutine)

asyncio.run(main())

7.8.4.4 - Asynchronous Timer​

Creating a simple asynchronous timer to execute tasks at regular intervals.

async def timer(interval, task):
    while True:
        await asyncio.sleep(interval)
        await task()

async def print_time():
    print(f"Task executed at {time.strftime('%X')}")

async def main():
    await timer(5, print_time)  # Execute `print_time` every 5 seconds

asyncio.run(main())

7.8.4.5 - Chat Server with Asyncio​

Building an asynchronous chat server to handle multiple client connections.

import asyncio

async def handle_client(reader, writer):
    while True:
        data = await reader.read(100)
        if data:
            message = data.decode()
            print(f"Received: {message}")
            writer.write(data)
            await writer.drain()
        else:
            print("Closing connection")
            writer.close()
            break

async def main():
    server = await asyncio.start_server(
        handle_client, '127.0.0.1', 8888)
    async with server:
        await server.serve_forever()

asyncio.run(main())

7.8.5 - Structured Concurrency with TaskGroup​

Prior to Python 3.11, running several coroutines concurrently and reliably handling their failures required careful use of asyncio.gather() or manual task bookkeeping. asyncio.TaskGroup (added in 3.11) provides structured concurrency: it's an asynchronous context manager that tracks every task created inside it and waits for all of them to finish before the with block exits. If any task fails, the remaining tasks are cancelled automatically.

import asyncio

async def play_track(track):
    await asyncio.sleep(1)
    print(f"Finished: {track}")

async def play_album(name, tracks):
    async with asyncio.TaskGroup() as tg:
        for track in tracks:
            tg.create_task(play_track(track), name=track)
    print(f"Album '{name}' finished playing")

asyncio.run(play_album("Sundowning", ["Track 1", "Track 2", "Track 3"]))

If one or more tasks in the group raise an exception, TaskGroup cancels the remaining tasks and raises an ExceptionGroup containing all of the failures, which you handle with except* (see 2.6.7 - Exception Groups and except* in the Exception Handling chapter):

async def fetch(name, should_fail):
    if should_fail:
        raise ValueError(f"{name} failed")
    return name

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch("A", False))
            tg.create_task(fetch("B", True))
            tg.create_task(fetch("C", True))
    except* ValueError as eg:
        for error in eg.exceptions:
            print(f"Handled: {error}")

asyncio.run(main())

7.8.6 - Timeouts with asyncio.timeout()​

Also added in Python 3.11, asyncio.timeout() is an asynchronous context manager that cancels the code inside its block if it doesn't finish within the given number of seconds. It's the recommended replacement for calling asyncio.wait_for() directly:

import asyncio

async def slow_operation():
    await asyncio.sleep(5)
    return "done"

async def main():
    try:
        async with asyncio.timeout(2):
            result = await slow_operation()
    except TimeoutError:
        print("The operation timed out")

asyncio.run(main())

7.8.7 - Inspecting Running Tasks​

Python 3.14 adds a command-line tool for inspecting the asyncio tasks of an already-running process, without modifying its code. Given the process ID (PID) of a running Python program:

python -m asyncio ps PID       # flat table of tasks, their coroutine stacks, and what they're awaiting
python -m asyncio pstree PID   # the same information rendered as a hierarchical call tree

This is primarily a debugging tool for diagnosing programs that appear to hang — pstree in particular makes it easy to see which task a stuck program is blocked on and how its coroutines are chained together.