Practical Examples of Python Coroutine Libraries: asyncio, gevent, and trio

Coroutines are lightweight concurrency models that enable efficient asynchronous operations within a single thread, and Python offers several coroutine libraries such as asyncio , gevent , and trio .

1. Async I/O with asyncio

import asyncio

async def fetch_data(url):
    # Simulate async I/O
    await asyncio.sleep(1)
    return f"Data from {url}"

async def main():
    tasks = [
        fetch_data("https://api.example.com/data1"),
        fetch_data("https://api.example.com/data2"),
        fetch_data("https://api.example.com/data3")
    ]
    results = await asyncio.gather(*tasks)
    print(results)

asyncio.run(main())

2. Concurrent network requests with gevent

import gevent
import requests

def fetch_data(url):
    response = requests.get(url)
    return response.text

def main():
    urls = [
        "https://api.example.com/data1",
        "https://api.example.com/data2",
        "https://api.example.com/data3"
    ]
    jobs = [gevent.spawn(fetch_data, url) for url in urls]
    gevent.joinall(jobs)
    results = [job.value for job in jobs]
    print(results)

main()

3. Asynchronous file read/write with trio

import trio

async def read_file(file_name):
    async with trio.open_file(file_name, "r") as file:
        data = await file.read()
        return data

async def write_file(file_name, data):
    async with trio.open_file(file_name, "w") as file:
        await file.write(data)

async def main():
    await write_file("output.txt", "Hello, world!")
    data = await read_file("output.txt")
    print(data)

trio.run(main)

4. Concurrent tasks with asyncio

import asyncio

async def task(name):
    print(f"Task {name} started")
    await asyncio.sleep(1)
    print(f"Task {name} completed")

async def main():
    tasks = [task("A"), task("B"), task("C")]
    await asyncio.gather(*tasks)

asyncio.run(main())

5. Concurrent web crawler with gevent

import gevent
import requests
from gevent import monkey

monkey.patch_all()

def fetch_data(url):
    response = requests.get(url)
    return response.text

def main():
    urls = [
        "https://www.example.com/page1",
        "https://www.example.com/page2",
        "https://www.example.com/page3"
    ]
    jobs = [gevent.spawn(fetch_data, url) for url in urls]
    gevent.joinall(jobs)
    results = [job.value for job in jobs]
    print(results)

main()

6. Asynchronous TCP server with trio

import trio

async def handle_client(stream):
    data = await stream.receive_some(1024)
    await stream.send_all(data.upper())
    await stream.aclose()

async def main():
    await trio.serve_tcp(handle_client, port=8000)

trio.run(main)

7. Periodic task with asyncio

import asyncio

async def task():
    while True:
        print("Running task")
        await asyncio.sleep(1)

async def main():
    await asyncio.gather(task(), asyncio.sleep(5))

asyncio.run(main())

8. Concurrent database access with gevent

import gevent
from gevent import monkey
from gevent.pool import Pool
import psycopg2

monkey.patch_all()

def fetch_data(query):
    conn = psycopg2.connect("dbname=mydatabase user=myuser password=mypassword")
    cursor = conn.cursor()
    cursor.execute(query)
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    return results

def main():
    queries = [
        "SELECT * FROM table1",
        "SELECT * FROM table2",
        "SELECT * FROM table3"
    ]
    pool = Pool(10)
    results = pool.map(fetch_data, queries)
    print(results)

main()

9. Asynchronous WebSocket server with trio

import trio
from trio_web import serve_websocket, ConnectionClosed

async def handle_websocket(request):
    ws = await request.accept()
    try:
        while True:
            message = await ws.receive_message()
            await ws.send_message(message)
    except ConnectionClosed:
        pass

async def main():
    await serve_websocket(handle_websocket, "0.0.0.0", 8000, ssl_context=None)

trio.run(main)

10. Asynchronous task queue with asyncio

import asyncio
import random

async def worker(name, queue):
    while True:
        task = await queue.get()
        if task is None:
            break
        print(f"Worker {name} got task: {task}")
        await asyncio.sleep(random.random())

async def main():
    queue = asyncio.Queue()
    workers = [worker("A", queue), worker("B", queue), worker("C", queue)]
    tasks = [f"Task {i}" for i in range(10)]
    random.shuffle(tasks)
    for task in tasks:
        await queue.put(task)
    for _ in workers:
        await queue.put(None)
    await asyncio.gather(*workers)

asyncio.run(main())

These examples demonstrate how asyncio , gevent , and trio can be applied to various asynchronous programming scenarios, offering different concurrency models and tools suitable for diverse application needs.