NeuralBlitz 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);
}
}