Understanding Python Coroutines: Concepts, Implementation, and Practical Examples

Introduction

In Python, a coroutine is a special type of function that can pause its execution and later resume, enabling lightweight asynchronous programming and concurrent processing.

1. Basic Concepts of Coroutines

1.1 Definition

A coroutine is a function that can be suspended and resumed, typically using yield or await statements, and it preserves its state between calls.

1.2 Advantages

Lightweight: lower creation and context‑switch overhead compared to threads or processes.

Asynchronous programming: avoids callback hell and simplifies async flow.

Resource management: better handling of resources, reducing contention and deadlocks.

2. Implementing Coroutines in Python

2.1 Using async and await

Python 3.5 introduced the async and await keywords, making asynchronous code more readable and concise.

2.2 Example: async and await

import asyncio
async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)

async def main():
    print(f"started at {time.strftime('%X')}")
    await say_after(1, 'hello')
    await say_after(2, 'world')
    print(f"finished at {time.strftime('%X')}")

asyncio.run(main())

3. How Coroutines Work

3.1 async def defines a coroutine function

The async def syntax creates a coroutine function that can contain await expressions to pause execution.

3.2 await suspends execution

The await statement pauses the current coroutine until the awaited coroutine completes, then resumes.

3.3 asyncio library

The built‑in asyncio library provides an event loop, task scheduling, and other utilities to support coroutine programming.

4. Example: Concurrent HTTP Requests with Coroutines

Using aiohttp and asyncio to fetch multiple URLs concurrently:

import asyncio
import aiohttp

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

async def main():
    urls = [
        "https://api.example.com/data1",
        "https://api.example.com/data2",
        "https://api.example.com/data3",
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        responses = await asyncio.gather(*tasks)
        for response in responses:
            print(response)

asyncio.run(main())

5. Example: Resource Management with Coroutines

A simple resource manager that acquires and releases resources asynchronously:

import asyncio

class ResourceManager:
    def __init__(self):
        self.resources = {}

    async def acquire(self, resource_id):
        await asyncio.sleep(1)  # simulate acquisition time
        print(f"Acquired resource {resource_id}")
        self.resources[resource_id] = True

    async def release(self, resource_id):
        await asyncio.sleep(1)  # simulate release time
        print(f"Released resource {resource_id}")
        del self.resources[resource_id]

async def use_resource(manager, resource_id):
    await manager.acquire(resource_id)
    print(f"Using resource {resource_id}")
    await asyncio.sleep(2)  # simulate usage time
    await manager.release(resource_id)
    print(f"Released resource {resource_id}")

async def main():
    manager = ResourceManager()
    tasks = [use_resource(manager, i) for i in range(5)]
    await asyncio.gather(*tasks)

asyncio.run(main())

6. Summary

The article covered the definition of coroutines, their advantages, how to implement them in Python using async and await , the underlying mechanics of the asyncio library, and practical use cases such as concurrent HTTP requests and resource management.