Asyncio Advanced Patterns

Introduction

Advanced asyncio patterns for complex concurrent applications.

Semaphores and Bounded Semaphores

import asyncio

semaphore = asyncio.Semaphore(3)

async def limited_task(n):
    async with semaphore:
        print(f"Task {n} starting")
        await asyncio.sleep(1)
        print(f"Task {n} completed")
        return n ** 2

async def main():
    tasks = [limited_task(i) for i in range(10)]
    return await asyncio.gather(*tasks)

Queues

import asyncio

async def producer(queue):
    for i in range(5):
        await queue.put(i)
        print(f"Produced {i}")
    await queue.join()

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

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

Task Groups

import asyncio

async def fetch(url):
    return url

async def main():
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch(url)) for url in urls]
    
    results = [task.result() for task in tasks]

Practice Problems

1. Implement rate limiting with semaphore
2. Create producer-consumer with queues
3. Use timeout for async operations
4. Handle cancellations gracefully
5. Coordinate multiple tasks