Python Asyncio Advanced Patterns

Introduction

Asyncio is Python core async library, but many developers only know basic usage. This article explores advanced patterns for maximum performance.

1. Semaphore for Concurrency Control

import asyncio

class RateLimiter:
    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def __aenter__(self):
        await self.semaphore.acquire()
        return self

    async def __aexit__(self, *args):
        self.semaphore.release()

2. Queue Processing Pattern

class AsyncWorkerPool:
    def __init__(self, worker_count, processor):
        self.queue = asyncio.Queue()
        self.worker_count = worker_count
        self.processor = processor

3. Mixing CPU-Intensive Tasks

from concurrent.futures import ProcessPoolExecutor
process_pool = ProcessPoolExecutor()

async def hybrid_processing(items):
    loop = asyncio.get_running_loop()
    raw_data = await asyncio.gather(*[fetch_data(i) for i in items])
    futures = [loop.run_in_executor(process_pool, cpu_task, d) for d in raw_data]
    return await asyncio.gather(*futures)

Summary

Asyncio advanced patterns help you build high-performance applications.

Tool Recommendation: Try PySecToolkit - a Python security toolkit built on asyncio.

---

Originally published on WD Tech Blog