Async/Await & Concurrency
Introduction
Python's asyncio provides a way to write asynchronous code using async and await keywords. This is similar to JavaScript's Promises and async/await, but Python's approach is more explicit about asynchronous operations.
Callback Hell in Python occurs when using nested threading callbacks. Asyncio solves this by providing:
- Clean async/await syntax
- Built-in event loop
- Concurrent execution without complex threading
- Better error handling and stack traces
Benefits of async/await:
- Flattens nested callbacks
- Makes async code look synchronous
- Better error handling
- Improved performance for I/O-bound operations
Concepts
Asyncio Fundamentals
async defdefines a coroutine functionawaitpauses execution until the awaited operation completesasyncio.run()starts the event loopasyncio.gather()runs multiple coroutines concurrently
Async vs Threading
- Asyncio: Single-threaded, cooperative multitasking
- Threading: Multi-threaded, preemptive multitasking
- Asyncio is better for I/O-bound operations
- Threading is better for CPU-bound operations
Examples
Callback Hell with Threading
import threading
import time
def callback_hell_example():
"""Example of nested callbacks (callback hell)"""
def get_user(callback):
def worker():
time.sleep(1) # Simulate API delay
user = {"id": 1, "name": "Alice"}
callback(user)
threading.Thread(target=worker).start()
def get_orders(user_id, callback):
def worker():
time.sleep(1)
orders = [{"id": 101, "user_id": user_id, "amount": 99.99}]
callback(orders)
threading.Thread(target=worker).start()
def get_order_details(order_id, callback):
def worker():
time.sleep(1)
details = {"id": order_id, "items": ["Laptop", "Mouse"], "status": "shipped"}
callback(details)
threading.Thread(target=worker).start()
# Nested callbacks (hard to read and maintain)
get_user(lambda user:
get_orders(user["id"], lambda orders:
get_order_details(orders[0]["id"], lambda details:
print(f"Order details: {details}")
)
)
)
# Run callback hell example
print("Callback Hell Example:")
callback_hell_example()
time.sleep(4) # Wait for completion
Clean Async/Await Solution
import asyncio
import random
async def get_user():
"""Async function to get user data"""
await asyncio.sleep(1) # Simulate API delay
return {"id": 1, "name": "Alice", "email": "alice@example.com"}
async def get_orders(user_id):
"""Async function to get user orders"""
await asyncio.sleep(1)
return [
{"id": 101, "user_id": user_id, "amount": 99.99, "status": "completed"},
{"id": 102, "user_id": user_id, "amount": 149.99, "status": "pending"}
]
async def get_order_details(order_id):
"""Async function to get order details"""
await asyncio.sleep(1)
return {
"id": order_id,
"items": ["Laptop", "Wireless Mouse", "USB Cable"],
"shipping": "express",
"tracking": f"TRACK{order_id}"
}
async def async_solution():
"""Clean async/await solution"""
try:
# Sequential execution (like Promise chain)
user = await get_user()
print(f"User: {user['name']}")
orders = await get_orders(user["id"])
print(f"Found {len(orders)} orders")
if orders:
details = await get_order_details(orders[0]["id"])
print(f"First order details: {details}")
except Exception as e:
print(f"Error: {e}")
# Run async solution
print("\nAsync/Await Solution:")
asyncio.run(async_solution())
Concurrent Execution
async def fetch_data(url, delay=1):
"""Simulate fetching data from different APIs"""
print(f"Fetching from {url}...")
await asyncio.sleep(delay)
return {
"url": url,
"data": f"Data from {url}",
"timestamp": asyncio.get_event_loop().time()
}
async def concurrent_example():
"""Run multiple async operations concurrently"""
urls = [
"https://api.users.com",
"https://api.orders.com",
"https://api.products.com"
]
# Start all requests concurrently
start_time = asyncio.get_event_loop().time()
# Method 1: Using asyncio.gather()
results = await asyncio.gather(*[fetch_data(url) for url in urls])
end_time = asyncio.get_event_loop().time()
print(f"Concurrent execution took {end_time - start_time:.2f} seconds")
for result in results:
print(f" {result['url']}: {result['data']}")
asyncio.run(concurrent_example())
Error Handling in Async Code
async def risky_operation(should_fail=False):
"""Operation that might fail"""
await asyncio.sleep(0.5)
if should_fail:
raise ValueError("Something went wrong!")
return "Success!"
async def error_handling_example():
"""Demonstrate error handling in async code"""
tasks = [
risky_operation(False),
risky_operation(True), # This will fail
risky_operation(False),
]
# Method 1: Handle errors individually
results = await asyncio.gather(*tasks, return_exceptions=True)
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"Task {i} failed: {result}")
else:
print(f"Task {i} succeeded: {result}")
# Method 2: Try/except for individual operations
try:
result = await risky_operation(True)
print(f"Result: {result}")
except ValueError as e:
print(f"Caught error: {e}")
asyncio.run(error_handling_example())
Challenges
1. Promise resolves after 1s
Create an async function that returns "Hello World" after 1 second.
💡 Show Answer
import asyncio
async def hello_after_delay():
"""Return hello message after 1 second delay"""
await asyncio.sleep(1)
return "Hello World"
async def test_hello():
message = await hello_after_delay()
print(message)
# Run it
asyncio.run(test_hello())
# Alternative using asyncio.create_task()
async def hello_with_task():
task = asyncio.create_task(hello_after_delay())
result = await task
print(f"Task result: {result}")
asyncio.run(hello_with_task())
# Multiple ways to create and await
async def various_approaches():
# Method 1: Direct await
result1 = await hello_after_delay()
print(f"Direct: {result1}")
# Method 2: Using task
task = asyncio.create_task(hello_after_delay())
result2 = await task
print(f"Task: {result2}")
# Method 3: Using ensure_future (older style)
future = asyncio.ensure_future(hello_after_delay())
result3 = await future
print(f"Future: {result3}")
asyncio.run(various_approaches())
2. Reject after 2s
Write an async function that raises an exception after 2 seconds and handle it properly.
💡 Show Answer
import asyncio
async def failing_operation():
"""Async operation that fails after 2 seconds"""
await asyncio.sleep(2)
raise ValueError("Error: Failed")
async def test_error_handling():
"""Test different ways to handle async errors"""
try:
result = await failing_operation()
print(f"Unexpected success: {result}")
except ValueError as e:
print(f"Caught expected error: {e}")
except Exception as e:
print(f"Caught unexpected error: {e}")
# Run basic error handling
asyncio.run(test_error_handling())
# Advanced error handling with timeout
async def failing_with_timeout():
"""Handle errors with timeout"""
try:
# Set a timeout of 3 seconds
result = await asyncio.wait_for(failing_operation(), timeout=3.0)
print(f"Result: {result}")
except asyncio.TimeoutError:
print("Operation timed out!")
except ValueError as e:
print(f"Operation failed: {e}")
asyncio.run(failing_with_timeout())
# Error handling with multiple operations
async def multiple_operations_with_errors():
"""Handle errors in multiple concurrent operations"""
async def maybe_fail(delay, should_fail, operation_name):
await asyncio.sleep(delay)
if should_fail:
raise RuntimeError(f"{operation_name} failed")
return f"{operation_name} succeeded"
operations = [
maybe_fail(1, False, "Operation A"),
maybe_fail(2, True, "Operation B"), # This will fail
maybe_fail(1.5, False, "Operation C"),
]
# Gather with return_exceptions=True to handle errors gracefully
results = await asyncio.gather(*operations, return_exceptions=True)
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"Operation {i} failed: {result}")
else:
print(f"Operation {i}: {result}")
asyncio.run(multiple_operations_with_errors())
# Custom exception handling
class APIError(Exception):
def __init__(self, message, status_code=500):
self.message = message
self.status_code = status_code
super().__init__(self.message)
async def api_call_simulation(endpoint, should_fail=False):
"""Simulate API call that might fail"""
await asyncio.sleep(1)
if should_fail:
raise APIError("API server error", 500)
return {"endpoint": endpoint, "data": "response data"}
async def robust_error_handling():
"""Robust error handling with custom exceptions"""
endpoints = [
("users", False),
("orders", True), # Will fail
("products", False)
]
results = []
for endpoint, should_fail in endpoints:
try:
result = await api_call_simulation(endpoint, should_fail)
results.append({"endpoint": endpoint, "success": True, "data": result})
except APIError as e:
results.append({
"endpoint": endpoint,
"success": False,
"error": e.message,
"status_code": e.status_code
})
except Exception as e:
results.append({
"endpoint": endpoint,
"success": False,
"error": str(e)
})
print("API Results:")
for result in results:
if result["success"]:
print(f"✅ {result['endpoint']}: {result['data']}")
else:
print(f"❌ {result['endpoint']}: {result['error']}")
asyncio.run(robust_error_handling())
3. Chain transforms
Chain async operations to transform a number (square it, then double it, then add 10).
💡 Show Answer
import asyncio
async def square_async(n):
"""Square a number asynchronously"""
await asyncio.sleep(0.2) # Simulate processing time
result = n * n
print(f"Squared {n} = {result}")
return result
async def double_async(n):
"""Double a number asynchronously"""
await asyncio.sleep(0.2)
result = n * 2
print(f"Doubled {n} = {result}")
return result
async def add_ten_async(n):
"""Add 10 to a number asynchronously"""
await asyncio.sleep(0.2)
result = n + 10
print(f"Added 10 to {n} = {result}")
return result
async def chain_transforms_basic():
"""Basic chaining of async transformations"""
number = 5
print(f"Starting with: {number}")
# Chain operations sequentially
squared = await square_async(number)
doubled = await double_async(squared)
final = await add_ten_async(doubled)
print(f"Final result: {final}")
return final
# Run basic chaining
print("Basic Chaining:")
asyncio.run(chain_transforms_basic())
# More elegant chaining approach
async def chain_transforms_elegant():
"""Elegant chaining using intermediate variables"""
number = 3
print(f"\nStarting with: {number}")
result = number
result = await square_async(result) # 3 -> 9
result = await double_async(result) # 9 -> 18
result = await add_ten_async(result) # 18 -> 28
print(f"Final result: {result}")
return result
asyncio.run(chain_transforms_elegant())
# Functional approach with pipeline
async def pipeline_transform(value, *operations):
"""Apply a pipeline of async operations to a value"""
result = value
print(f"Pipeline starting with: {result}")
for i, operation in enumerate(operations):
result = await operation(result)
print(f"Step {i+1} result: {result}")
print(f"Pipeline complete: {result}")
return result
async def test_pipeline():
"""Test the pipeline approach"""
operations = [square_async, double_async, add_ten_async]
result = await pipeline_transform(4, *operations)
return result
print("\nPipeline Approach:")
asyncio.run(test_pipeline())
# Advanced: Parallel vs Sequential chaining
async def parallel_independent_transforms():
"""Transform multiple numbers in parallel"""
numbers = [2, 3, 4, 5]
print(f"\nTransforming {numbers} in parallel:")
# Apply same transformation pipeline to all numbers concurrently
tasks = [
pipeline_transform(num, square_async, double_async, add_ten_async)
for num in numbers
]
results = await asyncio.gather(*tasks)
print(f"All results: {results}")
return results
asyncio.run(parallel_independent_transforms())
# Chaining with error handling
async def safe_square(n):
"""Square with validation"""
await asyncio.sleep(0.1)
if not isinstance(n, (int, float)):
raise TypeError("Input must be a number")
if n < 0:
raise ValueError("Cannot square negative numbers in this example")
return n * n
async def safe_double(n):
"""Double with validation"""
await asyncio.sleep(0.1)
if n > 1000:
raise ValueError("Number too large to double")
return n * 2
async def safe_add_ten(n):
"""Add 10 with validation"""
await asyncio.sleep(0.1)
return n + 10
async def chain_with_error_handling():
"""Chain transformations with comprehensive error handling"""
test_values = [3, -2, "invalid", 30] # Mix of valid and invalid inputs
for value in test_values:
try:
print(f"\nProcessing: {value}")
result = value
result = await safe_square(result)
result = await safe_double(result)
result = await safe_add_ten(result)
print(f"✅ Success: {value} -> {result}")
except (TypeError, ValueError) as e:
print(f"❌ Error processing {value}: {e}")
except Exception as e:
print(f"❌ Unexpected error processing {value}: {e}")
asyncio.run(chain_with_error_handling())
# Method chaining style (more Pythonic)
class AsyncMath:
"""Async math operations with method chaining"""
def __init__(self, value):
self.value = value
async def square(self):
await asyncio.sleep(0.1)
self.value = self.value * self.value
print(f"Squared -> {self.value}")
return self
async def double(self):
await asyncio.sleep(0.1)
self.value = self.value * 2
print(f"Doubled -> {self.value}")
return self
async def add(self, n):
await asyncio.sleep(0.1)
self.value = self.value + n
print(f"Added {n} -> {self.value}")
return self
def result(self):
return self.value
async def method_chaining_example():
"""Example using method chaining"""
print("\nMethod Chaining Style:")
math = AsyncMath(6)
print(f"Starting with: {math.value}")
# Chain methods fluently
result = await (await (await math.square()).double()).add(10)
print(f"Final result: {result.result()}")
asyncio.run(method_chaining_example())
# Bonus: Create a decorator for async chaining
def async_chain(*operations):
"""Decorator to create an async chain of operations"""
async def chain_executor(initial_value):
result = initial_value
for operation in operations:
result = await operation(result)
return result
return chain_executor
# Usage of the decorator
@async_chain(square_async, double_async, add_ten_async)
async def transform_number(n):
pass # The decorator handles the implementation
async def test_decorator():
"""Test the async chain decorator"""
print("\nDecorator Approach:")
result = await transform_number(7)
print(f"Decorator result: {result}")
asyncio.run(test_decorator())
4. Timeout handling
Create async operations with timeout handling and cancellation.
💡 Show Answer
import asyncio
import random
async def slow_operation(operation_name, duration):
"""Simulate a slow operation"""
print(f"Starting {operation_name} (expected duration: {duration}s)")
try:
await asyncio.sleep(duration)
return f"{operation_name} completed successfully"
except asyncio.CancelledError:
print(f"{operation_name} was cancelled")
raise
async def timeout_example():
"""Basic timeout handling"""
try:
# This will timeout because operation takes 3s but timeout is 2s
result = await asyncio.wait_for(
slow_operation("Database query", 3),
timeout=2.0
)
print(f"Result: {result}")
except asyncio.TimeoutError:
print("Operation timed out!")
asyncio.run(timeout_example())
# Advanced timeout handling with multiple operations
async def multiple_operations_with_timeout():
"""Handle timeouts with multiple concurrent operations"""
operations = [
("Fast API call", 0.5),
("Slow database query", 3.0),
("Medium file operation", 1.5),
]
tasks = []
for name, duration in operations:
# Wrap each operation with a timeout
task = asyncio.create_task(
asyncio.wait_for(slow_operation(name, duration), timeout=2.0)
)
tasks.append((name, task))
# Process results as they complete
for name, task in tasks:
try:
result = await task
print(f"✅ {name}: {result}")
except asyncio.TimeoutError:
print(f"⏰ {name}: Timed out")
except asyncio.CancelledError:
print(f"❌ {name}: Cancelled")
asyncio.run(multiple_operations_with_timeout())
# Graceful cancellation
async def cancellable_operation(name, total_steps=10):
"""Operation that can be cancelled gracefully"""
try:
for step in range(total_steps):
print(f"{name}: Step {step + 1}/{total_steps}")
await asyncio.sleep(0.5) # Simulate work
return f"{name} completed all {total_steps} steps"
except asyncio.CancelledError:
print(f"{name}: Gracefully cancelled at step {step + 1}")
# Cleanup code would go here
raise
async def cancellation_example():
"""Example of manual cancellation"""
task = asyncio.create_task(cancellable_operation("Data processing", 8))
# Let it run for a bit, then cancel
await asyncio.sleep(2.5) # Let it complete ~5 steps
task.cancel()
try:
result = await task
print(f"Result: {result}")
except asyncio.CancelledError:
print("Task was cancelled as expected")
asyncio.run(cancellation_example())
# Sophisticated timeout with retry mechanism
async def operation_with_retry(name, duration, fail_probability=0.3):
"""Operation that might fail and supports retry"""
await asyncio.sleep(duration)
if random.random() < fail_probability:
raise RuntimeError(f"{name} failed randomly")
return f"{name} succeeded"
async def retry_with_timeout(operation, max_retries=3, timeout_per_attempt=2.0):
"""Retry an operation with timeout per attempt"""
for attempt in range(max_retries):
try:
print(f"Attempt {attempt + 1}/{max_retries}")
result = await asyncio.wait_for(operation(), timeout=timeout_per_attempt)
return result
except asyncio.TimeoutError:
print(f"Attempt {attempt + 1} timed out")
except Exception as e:
print(f"Attempt {attempt + 1} failed: {e}")
if attempt < max_retries - 1:
print("Retrying...")
await asyncio.sleep(0.5) # Brief delay before retry
raise RuntimeError(f"All {max_retries} attempts failed")
async def test_retry_with_timeout():
"""Test retry mechanism with timeout"""
try:
result = await retry_with_timeout(
lambda: operation_with_retry("Critical API call", 1.0, 0.5),
max_retries=3,
timeout_per_attempt=1.5
)
print(f"✅ Final success: {result}")
except RuntimeError as e:
print(f"❌ Final failure: {e}")
asyncio.run(test_retry_with_timeout())
# Context manager for timeout handling
class AsyncTimeout:
"""Context manager for timeout handling"""
def __init__(self, timeout):
self.timeout = timeout
self.task = None
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.task and not self.task.done():
self.task.cancel()
try:
await self.task
except asyncio.CancelledError:
pass
async def run(self, coro):
"""Run coroutine with timeout"""
self.task = asyncio.create_task(coro)
try:
return await asyncio.wait_for(self.task, timeout=self.timeout)
except asyncio.TimeoutError:
print("Operation timed out in context manager")
raise
async def test_timeout_context_manager():
"""Test custom timeout context manager"""
async with AsyncTimeout(1.5) as timeout_mgr:
try:
result = await timeout_mgr.run(slow_operation("Context managed op", 2.0))
print(f"Result: {result}")
except asyncio.TimeoutError:
print("Handled timeout in context manager")
asyncio.run(test_timeout_context_manager())
5. Concurrent data processing
Create a system that processes multiple data sources concurrently with proper error handling.
💡 Show Answer
import asyncio
import aiohttp
import random
import json
from typing import List, Dict, Any
from dataclasses import dataclass
from datetime import datetime
@dataclass
class DataSource:
name: str
url: str
timeout: float = 5.0
retries: int = 2
@dataclass
class ProcessingResult:
source: str
success: bool
data: Any = None
error: str = None
duration: float = 0.0
timestamp: datetime = None
async def simulate_data_fetch(source: DataSource) -> Dict[str, Any]:
"""Simulate fetching data from various sources"""
start_time = asyncio.get_event_loop().time()
# Simulate different response times and failure rates
delays = {
"users_api": (0.5, 0.1), # 0.5s delay, 10% failure rate
"orders_db": (1.2, 0.2), # 1.2s delay, 20% failure rate
"inventory_service": (0.8, 0.15), # 0.8s delay, 15% failure rate
"analytics_api": (2.0, 0.25), # 2.0s delay, 25% failure rate
"payment_gateway": (1.5, 0.3), # 1.5s delay, 30% failure rate
}
delay, failure_rate = delays.get(source.name, (1.0, 0.2))
await asyncio.sleep(delay)
if random.random() < failure_rate:
raise ConnectionError(f"Failed to connect to {source.name}")
# Generate mock data based on source type
mock_data = {
"users_api": {"users": [{"id": i, "name": f"User{i}"} for i in range(1, 6)]},
"orders_db": {"orders": [{"id": i, "amount": round(random.uniform(10, 500), 2)} for i in range(1, 4)]},
"inventory_service": {"inventory": [{"item": f"Item{i}", "stock": random.randint(0, 100)} for i in range(1, 8)]},
"analytics_api": {"metrics": {"daily_active_users": random.randint(1000, 5000), "revenue": round(random.uniform(1000, 10000), 2)}},
"payment_gateway": {"transactions": [{"id": f"tx_{i}", "status": "completed"} for i in range(1, 3)]}
}
duration = asyncio.get_event_loop().time() - start_time
return {
"source": source.name,
"data": mock_data.get(source.name, {"message": "Generic data"}),
"fetch_time": duration,
"timestamp": datetime.now().isoformat()
}
async def fetch_with_retry(source: DataSource) -> ProcessingResult:
"""Fetch data with retry logic and timeout"""
start_time = asyncio.get_event_loop().time()
for attempt in range(source.retries + 1):
try:
print(f"📡 Fetching from {source.name} (attempt {attempt + 1})")
data = await asyncio.wait_for(
simulate_data_fetch(source),
timeout=source.timeout
)
duration = asyncio.get_event_loop().time() - start_time
return ProcessingResult(
source=source.name,
success=True,
data=data,
duration=duration,
timestamp=datetime.now()
)
except asyncio.TimeoutError:
error_msg = f"Timeout after {source.timeout}s"
print(f"⏰ {source.name}: {error_msg}")
if attempt == source.retries:
duration = asyncio.get_event_loop().time() - start_time
return ProcessingResult(
source=source.name,
success=False,
error=error_msg,
duration=duration,
timestamp=datetime.now()
)
except Exception as e:
error_msg = str(e)
print(f"❌ {source.name}: {error_msg}")
if attempt == source.retries:
duration = asyncio.get_event_loop().time() - start_time
return ProcessingResult(
source=source.name,
success=False,
error=error_msg,
duration=duration,
timestamp=datetime.now()
)
if attempt < source.retries:
wait_time = 2 ** attempt # Exponential backoff
print(f"🔄 Retrying {source.name} in {wait_time}s...")
await asyncio.sleep(wait_time)
async def process_data_sources(sources: List[DataSource]) -> List[ProcessingResult]:
"""Process multiple data sources concurrently"""
print(f"🚀 Starting concurrent processing of {len(sources)} sources...")
# Create tasks for all sources
tasks = [fetch_with_retry(source) for source in sources]
# Wait for all tasks to complete
results = await asyncio.gather(*tasks, return_exceptions=True)
# Handle any unexpected exceptions
processed_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
processed_results.append(ProcessingResult(
source=sources[i].name,
success=False,
error=f"Unexpected error: {result}",
timestamp=datetime.now()
))
else:
processed_results.append(result)
return processed_results
async def process_with_progress_tracking(sources: List[DataSource]):
"""Process sources with real-time progress tracking"""
completed = 0
total = len(sources)
results = []
async def track_progress(source: DataSource):
nonlocal completed
result = await fetch_with_retry(source)
completed += 1
progress = (completed / total) * 100
status = "✅" if result.success else "❌"
print(f"{status} Progress: {progress:.1f}% ({completed}/{total}) - {source.name}")
return result
# Create tasks with progress tracking
tasks = [track_progress(source) for source in sources]
results = await asyncio.gather(*tasks)
return results
async def main_concurrent_processing():
"""Main function demonstrating concurrent data processing"""
# Define data sources
sources = [
DataSource("users_api", "https://api.users.com", timeout=3.0, retries=2),
DataSource("orders_db", "postgresql://orders.db", timeout=4.0, retries=1),
DataSource("inventory_service", "http://inventory.internal", timeout=2.5, retries=3),
DataSource("analytics_api", "https://analytics.service", timeout=5.0, retries=1),
DataSource("payment_gateway", "https://payments.gateway", timeout=3.5, retries=2),
]
print("=" * 60)
print("CONCURRENT DATA PROCESSING EXAMPLE")
print("=" * 60)
start_time = asyncio.get_event_loop().time()
# Process all sources concurrently with progress tracking
results = await process_with_progress_tracking(sources)
end_time = asyncio.get_event_loop().time()
total_duration = end_time - start_time
# Analyze results
successful = [r for r in results if r.success]
failed = [r for r in results if not r.success]
print("\n" + "=" * 60)
print("PROCESSING SUMMARY")
print("=" * 60)
print(f"Total time: {total_duration:.2f}s")
print(f"Successful: {len(successful)}/{len(results)}")
print(f"Failed: {len(failed)}/{len(results)}")
if successful:
print("\n✅ SUCCESSFUL OPERATIONS:")
for result in successful:
print(f" {result.source}: {result.duration:.2f}s")
if result.data and 'data' in result.data:
data_summary = str(result.data['data'])[:50] + "..." if len(str(result.data['data'])) > 50 else str(result.data['data'])
print(f" Data: {data_summary}")
if failed:
print("\n❌ FAILED OPERATIONS:")
for result in failed:
print(f" {result.source}: {result.error}")
# Calculate efficiency
avg_duration = sum(r.duration for r in results) / len(results)
print(f"\nAverage operation time: {avg_duration:.2f}s")
# Demonstrate data aggregation
if successful:
print("\n📊 DATA AGGREGATION:")
aggregated_data = {}
for result in successful:
if result.data and 'data' in result.data:
aggregated_data[result.source] = result.data['data']
print(f"Combined data from {len(aggregated_data)} sources:")
for source, data in aggregated_data.items():
item_count = len(data) if isinstance(data, (list, dict)) else 1
print(f" {source}: {item_count} items")
# Run the concurrent processing example
asyncio.run(main_concurrent_processing())
# Bonus: Rate-limited concurrent processing
class RateLimiter:
"""Simple rate limiter for API calls"""
def __init__(self, max_calls: int, time_window: float):
self.max_calls = max_calls
self.time_window = time_window
self.calls = []
async def acquire(self):
"""Acquire permission to make a call"""
now = asyncio.get_event_loop().time()
# Remove calls outside the time window
self.calls = [call_time for call_time in self.calls if now - call_time < self.time_window]
# If we're at the limit, wait
if len(self.calls) >= self.max_calls:
wait_time = self.time_window - (now - self.calls[0])
if wait_time > 0:
print(f"Rate limit reached, waiting {wait_time:.2f}s...")
await asyncio.sleep(wait_time)
self.calls.append(now)
async def rate_limited_processing():
"""Example with rate limiting"""
print("\n" + "=" * 60)
print("RATE-LIMITED PROCESSING EXAMPLE")
print("=" * 60)
# Create rate limiter: max 3 calls per 2 seconds
rate_limiter = RateLimiter(max_calls=3, time_window=2.0)
async def rate_limited_fetch(source: DataSource):
await rate_limiter.acquire()
return await fetch_with_retry(source)
sources = [DataSource(f"api_service_{i}", f"https://api{i}.com") for i in range(6)]
start_time = asyncio.get_event_loop().time()
tasks = [rate_limited_fetch(source) for source in sources]
results = await asyncio.gather(*tasks)
total_time = asyncio.get_event_loop().time() - start_time
print(f"Rate-limited processing completed in {total_time:.2f}s")
successful = sum(1 for r in results if r.success)
print(f"Results: {successful}/{len(results)} successful")
# Uncomment to run rate-limited example
# asyncio.run(rate_limited_processing())