async-programming

What I do

• Write async/await code in Python and JavaScript
• Implement concurrent operations
• Handle async errors properly
• Use thread pools and process pools
• Implement producer-consumer patterns
• Manage async resources with context managers
• Handle backpressure and rate limiting
• Debug async code

When to use me

When implementing asynchronous operations or concurrency patterns.

Python Asyncio

import asyncio
import aiohttp
from typing import List, Optional
from dataclasses import dataclass
from contextlib import asynccontextmanager


@dataclass
class FetchResult:
    url: str
    status: int
    content: str
    duration_ms: float


class AsyncFetcher:
    """Concurrent HTTP fetcher with rate limiting."""
    
    def __init__(
        self,
        max_concurrent: int = 10,
        max_per_second: float = 5.0
    ) -> None:
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.rate_limiter = AsyncRateLimiter(max_per_second)
        self.results: List[FetchResult] = []
    
    async def fetch(
        self,
        session: aiohttp.ClientSession,
        url: str,
        timeout: int = 10
    ) -> FetchResult:
        """Fetch a single URL with rate limiting."""
        async with self.semaphore:
            async with self.rate_limiter:
                start = asyncio.get_event_loop().time()
                
                try:
                    async with session.get(
                        url,
                        timeout=aiohttp.ClientTimeout(total=timeout)
                    ) as response:
                        content = await response.text()
                        duration = (asyncio.get_event_loop().time() - start) * 1000
                        
                        return FetchResult(
                            url=url,
                            status=response.status,
                            content=content,
                            duration_ms=duration,
                        )
                except Exception as e:
                    return FetchResult(
                        url=url,
                        status=0,
                        content=str(e),
                        duration_ms=0,
                    )
    
    async def fetch_all(
        self,
        urls: List[str],
        progress_callback: Optional[callable] = None
    ) -> List[FetchResult]:
        """Fetch multiple URLs concurrently."""
        async with aiohttp.ClientSession() as session:
            tasks = [self.fetch(session, url) for url in urls]
            
            results = []
            for i, coro in enumerate(asyncio.as_completed(tasks)):
                result = await coro
                results.append(result)
                
                if progress_callback:
                    progress_callback(i + 1, len(urls))
            
            return results


class AsyncRateLimiter:
    """Token bucket rate limiter for async operations."""
    
    def __init__(
        self,
        rate: float,  # tokens per second
        capacity: Optional[int] = None
    ) -> None:
        self.rate = rate
        self.capacity = capacity or int(rate)
        self.tokens = self.capacity
        self.last_update = asyncio.get_event_loop().time()
        self.lock = asyncio.Lock()
    
    async def acquire(self, tokens: int = 1) -> float:
        """Acquire tokens, wait if necessary. Returns wait time."""
        async with self.lock:
            now = asyncio.get_event_loop().time()
            elapsed = now - self.last_update
            
            # Add tokens based on elapsed time
            new_tokens = elapsed * self.rate
            self.tokens = min(self.capacity, self.tokens + new_tokens)
            self.last_update = now
            
            # Wait for tokens if needed
            if self.tokens >= tokens:
                self.tokens -= tokens
                return 0.0
            
            # Calculate wait time
            needed = tokens - self.tokens
            wait_time = needed / self.rate
            self.tokens = 0
            self.last_update = now + wait_time
            
            return wait_time
    
    async def __aenter__(self) -> 'AsyncRateLimiter':
        await self.acquire()
        return self
    
    async def __aexit__(self, *args) -> None:
        pass


@asynccontextmanager
async def async_timer(name: str):
    """Context manager for timing async operations."""
    start = asyncio.get_event_loop().time()
    try:
        yield
    finally:
        duration = (asyncio.get_event_loop().time() - start) * 1000
        print(f"{name}: {duration:.2f}ms")

Concurrent Processing

import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import List, TypeVar, Callable, Awaitable


T = TypeVar('T')
R = TypeVar('R')


class ConcurrentProcessor:
    """Process items with configurable concurrency."""
    
    def __init__(
        self,
        max_workers: int = 10,
        use_processes: bool = False
    ) -> None:
        self.max_workers = max_workers
        self.executor = None
        self.use_processes = use_processes
    
    def __enter__(self) -> 'ConcurrentProcessor':
        if self.use_processes:
            self.executor = ProcessPoolExecutor(max_workers=self.max_workers)
        else:
            self.executor = ThreadPoolExecutor(max_workers=self.max_workers)
        return self
    
    def __exit__(self, *args) -> None:
        self.executor.shutdown(wait=True)
    
    async def map(
        self,
        func: Callable[[T], R],
        items: List[T],
        progress_callback: Optional[Callable[[int], None]] = None
    ) -> List[R]:
        """Map function over items concurrently."""
        loop = asyncio.get_event_loop()
        tasks = []
        
        for item in items:
            task = loop.run_in_executor(
                self.executor,
                func,
                item
            )
            tasks.append(task)
        
        results = []
        for i, future in enumerate(asyncio.as_completed(tasks)):
            result = await future
            results.append(result)
            if progress_callback:
                progress_callback(i + 1)
        
        return results
    
    async def map_ordered(
        self,
        func: Callable[[T], Awaitable[R]],
        items: List[T]
    ) -> List[R]:
        """Map async function preserving order."""
        tasks = [func(item) for item in items]
        return await asyncio.gather(*tasks)


class ProducerConsumer:
    """Producer-consumer pattern with bounded queue."""
    
    def __init__(
        self,
        max_queue_size: int = 100,
        max_workers: int = 5
    ) -> None:
        self.queue = asyncio.Queue(max_queue_size)
        self.workers = max_workers
        self.running = True
    
    async def producer(self, items: List[T]) -> None:
        """Produce items for consumption."""
        for item in items:
            await self.queue.put(item)
        
        # Signal completion
        for _ in range(self.workers):
            await self.queue.put(None)
    
    async def consumer(self, func: Callable[[T], Awaitable[R]]) -> List[R]:
        """Consume items and process them."""
        results = []
        
        while self.running:
            item = await self.queue.get()
            
            if item is None:
                self.queue.task_done()
                break
            
            try:
                result = await func(item)
                results.append(result)
            except Exception as e:
                print(f"Error processing item: {e}")
            finally:
                self.queue.task_done()
        
        return results
    
    async def run(
        self,
        items: List[T],
        func: Callable[[T], Awaitable[R]],
        progress_callback: Optional[Callable[[int], None]] = None
    ) -> List[R]:
        """Run the producer-consumer pipeline."""
        results = []
        
        async def worker():
            while True:
                item = await self.queue.get()
                if item is None:
                    break
                try:
                    result = await func(item)
                    results.append(result)
                finally:
                    self.queue.task_done()
        
        # Start workers
        workers = [worker() for _ in range(self.workers)]
        
        # Start producer
        producer_task = asyncio.create_task(self.producer(items))
        
        # Wait for all
        await asyncio.gather(producer_task, *workers)
        
        return results

Error Handling

import asyncio
from typing import Optional, TypeVar, Callable
from dataclasses import dataclass


@dataclass
class AsyncResult:
    success: bool
    value: Optional[R] = None
    error: Optional[Exception] = None
    duration_ms: float = 0


class AsyncErrorHandler:
    """Handle errors in async code with retries."""
    
    def __init__(
        self,
        max_retries: int = 3,
        backoff_factor: float = 1.0,
        exceptions: tuple = (Exception,)
    ) -> None:
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor
        self.exceptions = exceptions
    
    async def execute(
        self,
        func: Callable[..., Awaitable[R]],
        *args,
        **kwargs
    ) -> AsyncResult[R]:
        """Execute with retry logic."""
        start = asyncio.get_event_loop().time()
        
        for attempt in range(self.max_retries + 1):
            try:
                result = await func(*args, **kwargs)
                duration = (asyncio.get_event_loop().time() - start) * 1000
                
                return AsyncResult(
                    success=True,
                    value=result,
                    duration_ms=duration,
                )
            
            except self.exceptions as e:
                if attempt == self.max_retries:
                    duration = (asyncio.get_event_loop().time() - start) * 1000
                    return AsyncResult(
                        success=False,
                        error=e,
                        duration_ms=duration,
                    )
                
                delay = self.backoff_factor * (2 ** attempt)
                await asyncio.sleep(delay)
        
        duration = (asyncio.get_event_loop().time() - start) * 1000
        return AsyncResult(success=False, duration_ms=duration)


class TimeoutHandler:
    """Handle async operations with timeouts."""
    
    async def with_timeout(
        self,
        coro: Awaitable[R],
        timeout: float,
        fallback: Optional[Callable[[], R]] = None
    ) -> R:
        """Execute with timeout."""
        try:
            return await asyncio.wait_for(coro, timeout=timeout)
        except asyncio.TimeoutError:
            if fallback:
                return fallback()
            raise TimeoutError(f"Operation timed out after {timeout}s")
    
    async def with_individual_timeout(
        self,
        tasks: List[asyncio.Task],
        timeout: float
    ) -> List[Optional[R]]:
        """Wait for tasks with individual timeouts."""
        results = [None] * len(tasks)
        
        for i, task in enumerate(tasks):
            try:
                results[i] = await asyncio.wait_for(task, timeout=timeout)
            except asyncio.TimeoutError:
                results[i] = None
            except Exception:
                results[i] = None
        
        return results

JavaScript Async

// Promise utilities
class AsyncUtils {
  static async withTimeout<T>(
    promise: Promise<T>,
    timeoutMs: number,
    fallback?: T
  ): Promise<T> {
    return Promise.race([
      promise,
      new Promise<T>((_, reject) =>
        setTimeout(() => {
          if (fallback !== undefined) {
            resolve(fallback);
          } else {
            reject(new Error(`Timeout after ${timeoutMs}ms`));
          }
        }, timeoutMs)
      ),
    ]);
  }

  static async retry<T>(
    fn: () => Promise<T>,
    maxRetries: number = 3,
    delayMs: number = 1000
  ): Promise<T> {
    let lastError: Error | null = null;

    for (let i = 0; i <= maxRetries; i++) {
      try {
        return await fn();
      } catch (e) {
        lastError = e as Error;
        if (i < maxRetries) {
          await new Promise(resolve => setTimeout(resolve, delayMs * (i + 1)));
        }
      }
    }

    throw lastError;
  }

  static async allSettled<T>(
    promises: Promise<T>[]
  ): Promise<{ status: 'fulfilled' | 'rejected'; value?: T; reason?: Error }[]> {
    return Promise.all(
      promises.map(promise =>
        promise
          .then(value => ({ status: 'fulfilled' as const, value }))
          .catch(reason => ({ status: 'rejected' as const, reason }))
      )
    );
  }

  static async mapWithLimit<T, R>(
    items: T[],
    concurrency: number,
    fn: (item: T) => Promise<R>
  ): Promise<R[]> {
    const results: R[] = [];
    const executing: Promise<void>[] = [];

    for (const item of items) {
      const promise = Promise.resolve().then(() => fn(item));
      results.push(promise);
      executing.push(promise);

      if (executing.length >= concurrency) {
        await Promise.race(executing);
        const completed = executing.findIndex(p => p === promise);
        if (completed > -1) {
          executing.splice(completed, 1);
        }
      }
    }

    return Promise.all(results);
  }
}