async-programming

name	async-programming
description	Asynchronous programming patterns and best practices
license	MIT
compatibility	opencode
metadata	{"audience":"developers","category":"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() -> [R]:
        
        loop = asyncio.get_event_loop()
        tasks = []
        
         item  items:
            task = loop.run_in_executor(
                .executor,
                func,
                item
            )
            tasks.append(task)
        
        results = []
         i, future  (asyncio.as_completed(tasks)):
            result =  future
            results.append(result)
             progress_callback:
                progress_callback(i + )
        
         results
    
      () -> [R]:
        
        tasks = [func(item)  item  items]
          asyncio.gather(*tasks)


 :
    
    
     () -> :
        .queue = asyncio.Queue(max_queue_size)
        .workers = max_workers
        .running = 
    
      () -> :
        
         item  items:
             .queue.put(item)
        
        
         _  (.workers):
             .queue.put()
    
      () -> [R]:
        
        results = []
        
         .running:
            item =  .queue.get()
            
             item  :
                .queue.task_done()
                
            
            :
                result =  func(item)
                results.append(result)
             Exception  e:
                ()
            :
                .queue.task_done()
        
         results
    
      () -> [R]:
        
        results = []
        
          ():
             :
                item =  .queue.get()
                 item  :
                    
                :
                    result =  func(item)
                    results.append(result)
                :
                    .queue.task_done()
        
        
        workers = [worker()  _  (.workers)]
        
        
        producer_task = asyncio.create_task(.producer(items))
        
        
         asyncio.gather(producer_task, *workers)
        
         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) * 
                
                 AsyncResult(
                    success=,
                    value=result,
                    duration_ms=duration,
                )
            
             .exceptions  e:
                 attempt == .max_retries:
                    duration = (asyncio.get_event_loop().time() - start) * 
                     AsyncResult(
                        success=,
                        error=e,
                        duration_ms=duration,
                    )
                
                delay = .backoff_factor * ( ** attempt)
                 asyncio.sleep(delay)
        
        duration = (asyncio.get_event_loop().time() - start) * 
         AsyncResult(success=, duration_ms=duration)


 :
    
    
      () -> R:
        
        :
              asyncio.wait_for(coro, timeout=timeout)
         asyncio.TimeoutError:
             fallback:
                 fallback()
             TimeoutError()
    
      () -> [[R]]:
        
        results = [] * (tasks)
        
         i, task  (tasks):
            :
                results[i] =  asyncio.wait_for(task, timeout=timeout)
             asyncio.TimeoutError:
                results[i] = 
             Exception:
                results[i] = 
        
         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  ();
      }  (e) {
        lastError = e  ;
         (i < maxRetries) {
            ( (resolve, delayMs * (i + )));
        }
      }
    }

     lastError;
  }

    allSettled<T>(
    : <T>[]
  ): <{ :  | ; ?: T; ?:  }[]> {
     .(
      promises.(
        promise
          .( ({ :   , value }))
          .( ({ :   , reason }))
      )
    );
  }

    mapWithLimit<T, R>(
    : T[],
    : ,
    :  <R>
  ): <R[]> {
     : R[] = [];
     : <>[] = [];

     ( item  items) {
       promise = .().( (item));
      results.(promise);
      executing.(promise);

       (executing. >= concurrency) {
         .(executing);
         completed = executing.( p === promise);
         (completed > -) {
          executing.(completed, );
        }
      }
    }

     .(results);
  }
}