Agentient Async/Await Patterns with Comprehensive Error Handling
Core Principles
Async programming in Python 3.13 enables high-performance I/O-bound operations. All async code in Vibekit MUST include comprehensive error handling and proper resource cleanup.
Basic Async Functions
Every async function must handle errors properly:
import asyncio
from typing import Any
# ✅ REQUIRED: Comprehensive error handling in async functions
async def fetch_data(url: str, timeout: float = 10.0) -> dict[str, Any]:
"""
Fetch data from URL with timeout and error handling.
Raises:
TimeoutError: If request exceeds timeout
ConnectionError: If network request fails
ValueError: If response is not valid JSON
"""
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=timeout)) as response:
response.raise_for_status() # Raises on 4xx/5xx
return await response.json()
except asyncio.TimeoutError:
raise TimeoutError(f"Request to {url} timed out after {timeout}s")
except aiohttp.ClientError as e:
raise ConnectionError(f"Failed to fetch {url}: {e}")
except ValueError as e:
raise ValueError(f"Invalid JSON response from {url}: {e}")
# Let other exceptions propagate
# ❌ FORBIDDEN: Async function without error handling
async def bad_fetch(url: str):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.json() # No timeout, no error handling!
Concurrent Execution with asyncio.gather()
Run multiple async operations concurrently:
import asyncio
from typing import Any
# ✅ REQUIRED: Use asyncio.gather() for concurrent operations
async def fetch_multiple_urls(urls: list[str]) -> list[dict[str, Any]]:
"""
Fetch multiple URLs concurrently.
Returns list of results in same order as input URLs.
Failed requests return None instead of raising.
"""
tasks = [fetch_data(url) for url in urls]
# return_exceptions=True prevents one failure from cancelling all tasks
results = await asyncio.gather(*tasks, return_exceptions=True)
# Process results and exceptions
processed: list[dict[str, Any]] = []
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"Error fetching {urls[i]}: {result}")
processed.append(None)
else:
processed.append(result)
return processed
# Alternative: Fail-fast (any error cancels all)
async def fetch_multiple_strict(urls: list[str]) -> list[dict[str, Any]]:
"""Fetch URLs concurrently, raise on first error."""
tasks = [fetch_data(url) for url in urls]
return await asyncio.gather(*tasks) # return_exceptions=False (default)
Task Management with create_task()
Control task lifecycle explicitly:
import asyncio
from typing import Set
# ✅ REQUIRED: Track tasks and handle cleanup
class BackgroundTaskManager:
def __init__(self):
self.tasks: Set[asyncio.Task] = set()
def create_task(self, coro) -> asyncio.Task:
"""Create task and track it."""
task = asyncio.create_task(coro)
self.tasks.add(task)
task.add_done_callback(self.tasks.discard) # Auto-remove when done
return task
async def shutdown(self) -> None:
"""Cancel all pending tasks."""
for task in self.tasks:
task.cancel()
# Wait for all cancellations to complete
await asyncio.gather(*self.tasks, return_exceptions=True)
self.tasks.clear()
# Usage
async def main():
manager = BackgroundTaskManager()
# Start background tasks
task1 = manager.create_task(long_running_operation())
task2 = manager.create_task(periodic_cleanup())
try:
# Do main work
await some_main_task()
finally:
# Clean shutdown
await manager.shutdown()
Async Context Managers
Ensure proper resource cleanup in async code:
import asyncio
from typing import AsyncIterator
# ✅ REQUIRED: Use async context managers for resources
class AsyncDatabaseConnection:
"""Async context manager for database connections."""
def __init__(self, connection_string: str):
self.connection_string = connection_string
self.conn = None
async def __aenter__(self):
"""Establish connection on enter."""
try:
self.conn = await asyncio.wait_for(
connect_to_database(self.connection_string),
timeout=5.0
)
return self.conn
except asyncio.TimeoutError:
raise TimeoutError("Database connection timeout")
except Exception as e:
raise ConnectionError(f"Failed to connect: {e}")
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Close connection on exit, even if exception occurred."""
if self.conn is not None:
try:
await asyncio.wait_for(self.conn.close(), timeout=5.0)
except Exception as e:
print(f"Error closing connection: {e}")
return False # Don't suppress exceptions
# Usage
async def query_database():
async with AsyncDatabaseConnection("postgresql://...") as conn:
result = await conn.execute("SELECT * FROM users")
return result
# Alternative: Using contextlib.asynccontextmanager
from contextlib import asynccontextmanager
@asynccontextmanager
async def get_db_connection(conn_str: str) -> AsyncIterator[Any]:
"""Context manager for database connections."""
conn = None
try:
conn = await connect_to_database(conn_str)
yield conn
finally:
if conn:
await conn.close()
Timeouts and Cancellation
Always set timeouts for async operations:
import asyncio
from typing import Any
# ✅ REQUIRED: Use timeouts for all external I/O
async def fetch_with_timeout(url: str, timeout: float = 10.0) -> dict[str, Any]:
"""Fetch data with overall timeout."""
try:
return await asyncio.wait_for(
fetch_data(url),
timeout=timeout
)
except asyncio.TimeoutError:
raise TimeoutError(f"Operation timed out after {timeout}s")
# ✅ REQUIRED: Handle cancellation gracefully
async def long_running_task():
"""Task that handles cancellation properly."""
try:
for i in range(100):
await asyncio.sleep(1)
print(f"Step {i}")
except asyncio.CancelledError:
print("Task was cancelled, cleaning up...")
# Perform cleanup
raise # Re-raise to propagate cancellation
# ❌ FORBIDDEN: Async operation without timeout
async def bad_fetch():
return await fetch_data("https://slow-api.com") # No timeout!
Error Recovery with Retry Logic
Implement retry patterns for transient failures:
import asyncio
from typing import Any, Callable
# ✅ REQUIRED: Retry logic for transient errors
async def retry_async(
func: Callable,
*args,
max_attempts: int = 3,
delay: float = 1.0,
backoff: float = 2.0,
**kwargs
) -> Any:
"""
Retry async function with exponential backoff.
Args:
func: Async function to retry
max_attempts: Maximum number of attempts
delay: Initial delay between retries (seconds)
backoff: Multiplier for delay after each attempt
*args, **kwargs: Arguments to pass to func
Returns:
Result of func
Raises:
Last exception if all attempts fail
"""
last_exception = None
current_delay = delay
for attempt in range(max_attempts):
try:
return await func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt < max_attempts - 1:
print(f"Attempt {attempt + 1} failed: {e}. Retrying in {current_delay}s...")
await asyncio.sleep(current_delay)
current_delay *= backoff
else:
print(f"All {max_attempts} attempts failed")
raise last_exception
# Usage
async def unreliable_api_call(endpoint: str) -> dict:
return await fetch_data(endpoint)
result = await retry_async(
unreliable_api_call,
"https://api.example.com/data",
max_attempts=5,
delay=1.0,
backoff=2.0
)
Structured Concurrency with TaskGroup (Python 3.11+)
Use TaskGroup for managing related tasks:
import asyncio
# ✅ REQUIRED: Use TaskGroup for structured concurrency
async def fetch_all_data() -> list[dict]:
"""Fetch data from multiple sources using TaskGroup."""
results = []
async with asyncio.TaskGroup() as tg:
# All tasks started within the group
task1 = tg.create_task(fetch_data("https://api1.example.com"))
task2 = tg.create_task(fetch_data("https://api2.example.com"))
task3 = tg.create_task(fetch_data("https://api3.example.com"))
# At this point, all tasks have completed or an exception was raised
results = [task1.result(), task2.result(), task3.result()]
return results
# TaskGroup automatically:
# - Waits for all tasks to complete
# - Cancels remaining tasks if one raises an exception
# - Raises the first exception that occurred
Async Generators
Create async iterators for streaming data:
import asyncio
from typing import AsyncIterator
# ✅ REQUIRED: Async generators for streaming data
async def fetch_paginated_data(api_url: str) -> AsyncIterator[dict]:
"""Fetch paginated data, yielding each item."""
page = 1
while True:
try:
response = await fetch_data(f"{api_url}?page={page}")
items = response.get("items", [])
if not items:
break
for item in items:
yield item
page += 1
await asyncio.sleep(0.1) # Rate limiting
except Exception as e:
print(f"Error fetching page {page}: {e}")
break
# Usage
async def process_all_items():
async for item in fetch_paginated_data("https://api.example.com/items"):
await process_item(item)
Rate Limiting
Implement rate limiting for API calls:
import asyncio
import time
from typing import Any
class RateLimiter:
"""Rate limiter for async operations."""
def __init__(self, max_calls: int, period: float):
"""
Initialize rate limiter.
Args:
max_calls: Maximum number of calls allowed
period: Time period in seconds
"""
self.max_calls = max_calls
self.period = period
self.calls: list[float] = []
self.lock = asyncio.Lock()
async def acquire(self) -> None:
"""Wait until a call slot is available."""
async with self.lock:
now = time.time()
# Remove old calls outside the window
self.calls = [t for t in self.calls if now - t < self.period]
if len(self.calls) >= self.max_calls:
# Wait until the oldest call expires
sleep_time = self.period - (now - self.calls[0])
await asyncio.sleep(sleep_time)
self.calls.pop(0)
self.calls.append(time.time())
# Usage
limiter = RateLimiter(max_calls=10, period=60.0) # 10 calls per minute
async def rate_limited_fetch(url: str) -> dict:
await limiter.acquire()
return await fetch_data(url)
Anti-Patterns to Avoid
❌ No Error Handling in Async Functions
# BAD: Silent failures
async def bad_fetch(url: str):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.json()
# GOOD: Explicit error handling
async def good_fetch(url: str) -> dict:
try:
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
response.raise_for_status()
return await response.json()
except aiohttp.ClientError as e:
raise ConnectionError(f"Failed to fetch {url}: {e}")
❌ No Timeout on Async Operations
# BAD: Can hang forever
result = await fetch_data(url)
# GOOD: Always use timeout
result = await asyncio.wait_for(fetch_data(url), timeout=10.0)
❌ Not Re-raising CancelledError
# BAD: Suppresses cancellation
async def bad_task():
try:
await long_operation()
except asyncio.CancelledError:
print("Cancelled")
# Missing: raise
# GOOD: Re-raise to propagate cancellation
async def good_task():
try:
await long_operation()
except asyncio.CancelledError:
print("Cancelled, cleaning up...")
# Cleanup
raise # Re-raise
❌ Using Blocking Code in Async Functions
# BAD: Blocks the event loop
async def bad_async():
time.sleep(10) # Blocking!
return "done"
# GOOD: Use async sleep
async def good_async():
await asyncio.sleep(10) # Non-blocking
return "done"
❌ Not Tracking Background Tasks
# BAD: Task might be garbage collected
async def bad_background():
asyncio.create_task(background_work()) # Fire and forget - dangerous!
# GOOD: Track the task
async def good_background():
task = asyncio.create_task(background_work())
# Store task reference or await it later
return task
When to Use This Skill
Activate this skill when:
- Implementing async API clients or server handlers
- Creating concurrent data fetching operations
- Building async context managers
- Implementing retry logic for async operations
- Working with async generators or streaming data
- Managing background tasks with proper lifecycle
Related Resources
For additional patterns, see:
- Python asyncio Documentation: https://docs.python.org/3/library/asyncio.html
- Python 3.13 Patterns: See
python-3-13-patternsskill