Python Asyncio: Coroutines and Asynchronous Programming with Practical Code Examples

Coroutines and asynchronous programming are essential concepts in Python for handling concurrency and non‑blocking tasks; coroutines allow a program to pause and resume execution, while the async programming model builds on coroutines to boost concurrent performance using non‑blocking I/O.

Python's asynchronous capabilities are primarily provided by the asyncio module, which offers tools such as coroutines, an event loop, and asynchronous I/O operations.

1. Define a coroutine with async and await :

import asyncio
async def my_coroutine():
    await asyncio.sleep(1)
    print("Coroutine executed")
asyncio.run(my_coroutine())

2. Run multiple coroutines concurrently using asyncio.create_task() :

import asyncio
async def coroutine1():
    await asyncio.sleep(1)
    print("Coroutine 1 executed")
async def coroutine2():
    await asyncio.sleep(2)
    print("Coroutine 2 executed")
async def main():
    task1 = asyncio.create_task(coroutine1())
    task2 = asyncio.create_task(coroutine2())
    await asyncio.gather(task1, task2)
asyncio.run(main())

3. Wait for multiple coroutines with asyncio.wait() :

import asyncio
async def coroutine1():
    await asyncio.sleep(1)
    print("Coroutine 1 executed")
async def coroutine2():
    await asyncio.sleep(2)
    print("Coroutine 2 executed")
async def main():
    tasks = [coroutine1(), coroutine2()]
    done, pending = await asyncio.wait(tasks)
    for task in done:
        print(f"Task {task} completed")
asyncio.run(main())

4. Use asyncio.Lock() for mutual exclusion:

import asyncio
async def counter(lock):
    async with lock:
        for _ in range(5):
            print("Counting")
            await asyncio.sleep(1)
async def main():
    lock = asyncio.Lock()
    tasks = [counter(lock) for _ in range(3)]
    await asyncio.gather(*tasks)
asyncio.run(main())

5. Implement message passing with asyncio.Queue() :

import asyncio
async def producer(queue):
    for i in range(5):
        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}")
        await asyncio.sleep(2)
async def main():
    queue = asyncio.Queue()
    producer_task = asyncio.create_task(producer(queue))
    consumer_task = asyncio.create_task(consumer(queue))
    await asyncio.gather(producer_task, consumer_task)
asyncio.run(main())

6. Set a timeout for a coroutine using asyncio.wait_for() and handle TimeoutError :

import asyncio
async def my_coroutine():
    await asyncio.sleep(2)
    print("Coroutine executed")
async def main():
    try:
        await asyncio.wait_for(my_coroutine(), timeout=1)
    except asyncio.TimeoutError:
        print("Coroutine timed out")
asyncio.run(main())

7. Run blocking synchronous code in a coroutine with asyncio.run_in_executor() :

import asyncio
def sync_operation():
    # Blocking synchronous operation
    return "Sync result"
async def main():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, sync_operation)
    print(f"Result: {result}")
asyncio.run(main())

8. Perform asynchronous HTTP requests using the aiohttp library:

import asyncio
import aiohttp
async def fetch_data(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()
async def main():
    url = "https://api.example.com/data"
    data = await fetch_data(url)
    print(f"Data: {data}")
asyncio.run(main())

9. Simulate an asynchronous timer with asyncio.sleep() :

import asyncio
async def timer(duration):
    await asyncio.sleep(duration)
    print(f"Timer finished after {duration} seconds")
async def main():
    tasks = [timer(1), timer(2), timer(3)]
    await asyncio.gather(*tasks)
asyncio.run(main())

10. Implement concurrent file I/O using asyncio (illustrative example):

import asyncio
async def read_file(file):
    async with asyncio.open_file(file, "r") as f:
        contents = await f.read()
        print(f"Read from {file}: {contents}")
async def write_file(file, data):
    async with asyncio.open_file(file, "w") as f:
        await f.write(data)
        print(f"Wrote to {file}")
async def main():
    file = "data.txt"
    await write_file(file, "Hello, world!")
    await read_file(file)
asyncio.run(main())

These examples demonstrate how to use the asyncio module and related tools to write concurrent and asynchronous Python code, thereby improving program performance and responsiveness.