Asynchronous Programming in Python: Building High-Performance Applications

import requests
import time

def fetch_three_apis():
    start = time.time()
    
    requests.get('https://api1.example.com/data')  # Wait 1 second
    requests.get('https://api2.example.com/data')  # Wait 1 second  
    requests.get('https://api3.example.com/data')  # Wait 1 second
    
    print(f"Total time: {time.time() - start:.1f} seconds")  # ~3 seconds

import asyncio
import aiohttp

async def fetch_three_apis_async():
    async with aiohttp.ClientSession() as session:
        tasks = [
            session.get('https://api1.example.com/data'),
            session.get('https://api2.example.com/data'),
            session.get('https://api3.example.com/data')
        ]
        responses = await asyncio.gather(*tasks)
    # Total time: ~1 second (concurrent requests)

# Synchronous: 10 websites × 2 seconds each = 20 seconds
# Async: 10 websites concurrently = ~2 seconds

# Get user data, preferences, and recent activity simultaneously
user_data, preferences, activity = await asyncio.gather(
    fetch_user(user_id),
    fetch_preferences(user_id), 
    fetch_recent_activity(user_id)
)

# Process 100 log files concurrently instead of sequentially
# Reduces processing time from hours to minutes

# Fetch related data in parallel instead of sequential queries
# Reduces page load time from 500ms to 100ms

python -m venv async_env
source async_env/bin/activate  # Windows: async_env\Scripts\activate

pip install aiohttp aiofiles pytest-asyncio
pip freeze > requirements.txt

# src/config.py
import os
from dataclasses import dataclass

@dataclass
class Config:
    timeout: float = 30.0
    max_connections: int = 100
    debug: bool = os.getenv('DEBUG', 'False').lower() == 'true'

# src/http_client.py
import aiohttp

class AsyncHttpClient:
    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.session.close()
    
    async def get(self, url):
        async with self.session.get(url) as response:
            return await response.json()

# first_async.py
import asyncio
import time

async def make_coffee(name):
    print(f"☕ {name}: Starting to make coffee")
    await asyncio.sleep(3)  # Coffee takes 3 seconds
    print(f"☕ {name}: Coffee ready!")
    return f"{name}'s coffee"

async def make_toast(name):
    print(f"🍞 {name}: Starting to make toast")
    await asyncio.sleep(2)  # Toast takes 2 seconds  
    print(f"🍞 {name}: Toast ready!")
    return f"{name}'s toast"

async def make_breakfast():
    start_time = time.time()
    
    # Make coffee and toast simultaneously
    coffee, toast = await asyncio.gather(
        make_coffee("Alice"),
        make_toast("Alice")
    )
    
    elapsed = time.time() - start_time
    print(f"🍽️ Breakfast ready in {elapsed:.1f} seconds!")

asyncio.run(make_breakfast())

# tests/test_main.py
import pytest
import asyncio

@pytest.mark.asyncio
async def test_make_coffee():
    result = await make_coffee("Test")
    assert result == "Test's coffee"

@pytest.mark.asyncio  
async def test_concurrent_operations():
    start = time.time()
    
    # Run 5 tasks concurrently
    tasks = [asyncio.sleep(0.1) for _ in range(5)]
    await asyncio.gather(*tasks)
    
    elapsed = time.time() - start
    assert elapsed < 0.2  # Should take ~0.1s, not ~0.5s

pytest tests/

# Wrong - returns a coroutine object
result = async_function()

# Correct - waits for completion  
result = await async_function()

# Wrong - blocks the event loop
import time
time.sleep(1)

# Correct - yields to event loop
await asyncio.sleep(1)

# Wrong - might not close properly
session = aiohttp.ClientSession()
response = await session.get(url)

# Correct - always closes
async with aiohttp.ClientSession() as session:
    async with session.get(url) as response:
        data = await response.json()

import asyncio

async def say_hello(name, delay):
    """Async function that greets someone after a delay"""
    print(f"Hello {name}, starting...")
    await asyncio.sleep(delay)  # Non-blocking sleep
    print(f"Hello {name}, finished after {delay}s!")
    return f"Greeted {name}"

# Run a single async function
async def main():
    result = await say_hello("Alice", 2)
    print(f"Result: {result}")

# Execute the async program
asyncio.run(main())

import asyncio
import time

async def fetch_data(source, delay):
    """Simulate fetching data from different sources"""
    print(f"Fetching from {source}...")
    await asyncio.sleep(delay)  # Simulate network delay
    print(f"Got data from {source}")
    return f"Data from {source}"

async def sequential_example():
    """Sequential execution - slow"""
    print("=== Sequential Execution ===")
    start_time = time.time()
    
    result1 = await fetch_data("Database", 2)
    result2 = await fetch_data("API", 1.5)
    result3 = await fetch_data("Cache", 0.5)
    
    total_time = time.time() - start_time
    print(f"Sequential total time: {total_time:.1f}s\n")
    
    return [result1, result2, result3]

async def concurrent_example():
    """Concurrent execution - fast"""
    print("=== Concurrent Execution ===")
    start_time = time.time()
    
    # Start all operations simultaneously
    task1 = asyncio.create_task(fetch_data("Database", 2))
    task2 = asyncio.create_task(fetch_data("API", 1.5))
    task3 = asyncio.create_task(fetch_data("Cache", 0.5))
    
    # Wait for all to complete
    results = await asyncio.gather(task1, task2, task3)
    
    total_time = time.time() - start_time
    print(f"Concurrent total time: {total_time:.1f}s\n")
    
    return results

async def main():
    # Run both examples
    await sequential_example()
    await concurrent_example()

asyncio.run(main())

async def background_task(name):
    """Task that runs in background"""
    await asyncio.sleep(3)
    print(f"Background task {name} completed")

async def fire_and_forget_example():
    """Start tasks without waiting for them"""
    
    # Start background tasks
    asyncio.create_task(background_task("Task-1"))
    asyncio.create_task(background_task("Task-2"))
    
    # Do other work immediately
    print("Main work starting...")
    await asyncio.sleep(1)
    print("Main work finished")
    
    # Wait a bit to see background tasks complete
    await asyncio.sleep(3)

asyncio.run(fire_and_forget_example())

async def slow_operation():
    """Operation that might take too long"""
    await asyncio.sleep(5)
    return "Slow result"

async def timeout_example():
    """Handle operations with timeout"""
    try:
        # Wait maximum 2 seconds
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
        print(f"Got result: {result}")
    except asyncio.TimeoutError:
        print("Operation timed out!")

### Pattern 3: Error Handling in Concurrent Operations

```python
async def risky_operation(name, should_fail=False):
    """Operation that might fail"""
    await asyncio.sleep(1)
    if should_fail:
        raise ValueError(f"Operation {name} failed!")
    return f"Success from {name}"

async def error_handling_example():
    """Handle errors in concurrent operations"""
    tasks = [
        asyncio.create_task(risky_operation("Task-1", False)),
        asyncio.create_task(risky_operation("Task-2", True)),  # This will fail
        asyncio.create_task(risky_operation("Task-3", False))
    ]
    
    # Use return_exceptions=True to get both results and exceptions
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Task {i+1} failed: {result}")
        else:
            print(f"Task {i+1} succeeded: {result}")

asyncio.run(error_handling_example())

import asyncio
import aiohttp
import time

async def fetch_url(session, url):
    """Fetch a single URL"""
    try:
        async with session.get(url) as response:
            content = await response.text()
            return f"Success {url}: {len(content)} characters"
    except Exception as e:
        return f"Failed {url}: Error - {str(e)}"

async def scrape_websites():
    """Scrape multiple websites concurrently"""
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2", 
        "https://httpbin.org/delay/1",
        "https://httpbin.org/status/200",
        "https://httpbin.org/json"
    ]
    
    start_time = time.time()
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    
    total_time = time.time() - start_time
    
    print("Scraping Results:")
    for result in results:
        print(f"  {result}")
    print(f"\nTotal time: {total_time:.1f}s")
    print(f"Average per URL: {total_time/len(urls):.1f}s")

# Run the scraper
asyncio.run(scrape_websites())

# Wrong - creates coroutine but doesn't execute it
async def wrong_way():
    result = fetch_data("API", 1)  # Missing await!
    print(result)  # Prints: <coroutine object>

# Correct - actually executes the coroutine
async def right_way():
    result = await fetch_data("API", 1)  # With await
    print(result)  # Prints: "Data from API"

# Wrong - blocks the entire event loop
async def blocking_sleep():
    time.sleep(2)  # Blocks everything!

# Correct - allows other tasks to run
async def non_blocking_sleep():
    await asyncio.sleep(2)  # Other tasks can run

# Wrong - one failure stops everything
async def fragile_approach():
    results = await asyncio.gather(
        risky_operation("A"),
        risky_operation("B", should_fail=True),  # This stops everything
        risky_operation("C")
    )

# Correct - handle exceptions gracefully
async def robust_approach():
    results = await asyncio.gather(
        risky_operation("A"),
        risky_operation("B", should_fail=True),
        risky_operation("C"),
        return_exceptions=True  # Continue despite failures
    )

# Add this to see detailed async debugging info
import asyncio
asyncio.run(main(), debug=True)

async def debug_example():
    current = asyncio.current_task()
    print(f"Current task: {current.get_name()}")
    
    all_tasks = asyncio.all_tasks()
    print(f"Total running tasks: {len(all_tasks)}")

import asyncio

async def task_a():
    print("Task A: Starting")
    await asyncio.sleep(1)
    print("Task A: Finished")

async def task_b():
    print("Task B: Starting")
    await asyncio.sleep(0.5)
    print("Task B: Finished")

# The event loop coordinates both tasks
asyncio.run(asyncio.gather(task_a(), task_b()))

async def show_event_loop_work():
    loop = asyncio.get_running_loop()
    
    print(f"Loop is running: {loop.is_running()}")
    
    # Schedule a callback
    def callback():
        print("Callback executed!")
    
    loop.call_later(0.5, callback)
    await asyncio.sleep(1)  # Let the callback run

asyncio.run(show_event_loop_work())

async def background_work(name, duration):
    print(f"{name}: Starting work")
    await asyncio.sleep(duration)
    print(f"{name}: Work complete")
    return f"{name} result"

async def task_management_demo():
    # Create tasks explicitly
    task1 = asyncio.create_task(background_work("Worker-1", 2))
    task2 = asyncio.create_task(background_work("Worker-2", 1))
    
    print("Tasks created, doing other work...")
    await asyncio.sleep(0.5)
    
    # Check task status
    print(f"Task1 done: {task1.done()}")
    print(f"Task2 done: {task2.done()}")
    
    # Wait for completion
    results = await asyncio.gather(task1, task2)
    print(f"Results: {results}")

asyncio.run(task_management_demo())

import concurrent.futures
import time

def blocking_operation(duration):
    """Simulate CPU-intensive work"""
    time.sleep(duration)  # This blocks!
    return f"Blocked for {duration} seconds"

async def handle_blocking_work():
    loop = asyncio.get_running_loop()
    
    # Right way - run in thread pool
    with concurrent.futures.ThreadPoolExecutor() as executor:
        result = await loop.run_in_executor(
            executor, 
            blocking_operation, 
            2
        )
    
    print(f"Result: {result}")

asyncio.run(handle_blocking_work())

# Bad - blocks the event loop
async def bad_example():
    time.sleep(1)  # Blocks everything!
    return "Bad"

# Good - uses async sleep
async def good_example():
    await asyncio.sleep(1)  # Allows other tasks to run
    return "Good"

async def risky_task():
    await asyncio.sleep(0.1)
    raise ValueError("Something went wrong!")

async def exception_handling():
    # For multiple tasks, use return_exceptions
    tasks = [risky_task() for _ in range(3)]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Task {i} failed: {result}")

import asyncio
import time

async def monitor_event_loop():
    """Monitor event loop performance"""
    loop = asyncio.get_running_loop()
    
    # Enable debug mode for detailed info
    loop.set_debug(True)
    
    # Check current tasks
    all_tasks = asyncio.all_tasks(loop)
    print(f"Currently running tasks: {len(all_tasks)}")
    
    for task in all_tasks:
        print(f"  - {task.get_name()}: {task.get_coro()}")
    
    # Monitor loop time
    start_time = loop.time()
    await asyncio.sleep(0.1)
    elapsed = loop.time() - start_time
    print(f"Loop time elapsed: {elapsed:.3f}s")

async def slow_task():
    """Task that might slow down the loop"""
    print("Slow task starting...")
    # Simulate work that yields control frequently
    for i in range(5):
        await asyncio.sleep(0.1)  # Yield control
        print(f"  Slow task step {i+1}")
    print("Slow task finished")

async def debugging_demo():
    # Start monitoring
    monitor_task = asyncio.create_task(monitor_event_loop())
    slow_work = asyncio.create_task(slow_task())
    
    await asyncio.gather(monitor_task, slow_work)

asyncio.run(debugging_demo())

# Wrong - coroutines don't run until awaited
async def wrong_way():
    background_work("Task-1", 2)  # Just creates coroutine object
    background_work("Task-2", 1)  # Doesn't actually run
    await asyncio.sleep(3)

# Right - create tasks to run concurrently
async def right_way():
    task1 = asyncio.create_task(background_work("Task-1", 2))
    task2 = asyncio.create_task(background_work("Task-2", 1))
    await asyncio.sleep(3)  # Tasks run in background

import requests

# Wrong - blocks the entire event loop
async def blocking_http_request():
    response = requests.get("https://httpbin.org/delay/2")  # Blocks!
    return response.json()

# Right - use async HTTP client
import aiohttp

async def non_blocking_http_request():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://httpbin.org/delay/2") as response:
            return await response.json()

async def cancellable_task():
    """Task that handles cancellation gracefully"""
    try:
        for i in range(10):
            print(f"Working... step {i}")
            await asyncio.sleep(0.5)
    except asyncio.CancelledError:
        print("Task was cancelled, cleaning up...")
        # Perform cleanup here
        raise  # Re-raise to properly cancel

async def cancellation_demo():
    task = asyncio.create_task(cancellable_task())
    
    # Let it run for a bit
    await asyncio.sleep(2)
    
    # Cancel the task
    task.cancel()
    
    try:
        await task
    except asyncio.CancelledError:
        print("Task cancellation handled")

asyncio.run(cancellation_demo())

# Less efficient - many small operations
async def many_small_operations():
    results = []
    for i in range(100):
        result = await small_async_operation(i)
        results.append(result)
    return results

# More efficient - batch operations
async def batched_operations():
    tasks = [small_async_operation(i) for i in range(100)]
    return await asyncio.gather(*tasks)

async def small_async_operation(n):
    await asyncio.sleep(0.01)
    return n * 2

async def limited_concurrency_demo():
    """Limit concurrent operations to prevent overwhelming resources"""
    semaphore = asyncio.Semaphore(5)  # Max 5 concurrent operations
    
    async def limited_operation(n):
        async with semaphore:
            print(f"Starting operation {n}")
            await asyncio.sleep(1)
            print(f"Finished operation {n}")
            return n
    
    # Start 20 operations, but only 5 run at once
    tasks = [limited_operation(i) for i in range(20)]
    results = await asyncio.gather(*tasks)
    return results

asyncio.run(limited_concurrency_demo())

import asyncio

async def simple_coroutine():
    print("Coroutine started")
    await asyncio.sleep(1)  # Pause here
    print("Coroutine resumed")
    return "Done"

# Running a coroutine
result = asyncio.run(simple_coroutine())
print(result)  # "Done"

import inspect

async def example_coroutine():
    await asyncio.sleep(1)
    return "completed"

# Create but don't run the coroutine
coro = example_coroutine()
print(f"State: {inspect.getcoroutinestate(coro)}")  # CORO_CREATED

# Must run to see other states
asyncio.run(coro)

async def fetch_user(user_id):
    await asyncio.sleep(0.5)  # Simulate API call
    return {
        "id": user_id,
        "name": f"User {user_id}",
        "email": f"user{user_id}@example.com"
    }

async def process_user(user_id):
    user = await fetch_user(user_id)
    user["processed"] = True
    return user

async def main():
    # Process multiple users concurrently
    user_ids = [1, 2, 3]
    tasks = [process_user(uid) for uid in user_ids]
    users = await asyncio.gather(*tasks)
    
    for user in users:
        print(f"Processed: {user['name']}")

asyncio.run(main())

async def producer(queue, items):
    """Produce items and put them in queue"""
    for item in items:
        await queue.put(item)
        print(f"📦 Produced: {item}")
        await asyncio.sleep(0.1)
    
    await queue.put(None)  # Signal completion

async def consumer(queue, name):
    """Consume items from queue"""
    while True:
        item = await queue.get()
        
        if item is None:  # End signal
            break
        
        print(f"Processing {name}: {item}")
        await asyncio.sleep(0.2)  # Simulate processing

async def queue_demo():
    queue = asyncio.Queue(maxsize=2)  # Bounded queue
    
    items = ["task-1", "task-2", "task-3"]
    
    await asyncio.gather(
        producer(queue, items),
        consumer(queue, "Worker-A")
    )

asyncio.run(queue_demo())

class AsyncDatabase:
    def __init__(self, name):
        self.name = name
        self.connected = False
    
    async def __aenter__(self):
        print(f"🔌 Connecting to {self.name}")
        await asyncio.sleep(0.1)  # Simulate connection time
        self.connected = True
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print(f"🔌 Disconnecting from {self.name}")
        await asyncio.sleep(0.05)  # Simulate cleanup time
        self.connected = False
    
    async def query(self, sql):
        if not self.connected:
            raise RuntimeError("Database not connected")
        
        await asyncio.sleep(0.2)  # Simulate query time
        return f"Results for: {sql}"

async def database_operations():
    """Use database with proper cleanup"""
    async with AsyncDatabase("UserDB") as db:
        result = await db.query("SELECT * FROM users")
        return result

result = asyncio.run(database_operations())
print(f"Query result: {result}")

import random

async def unreliable_task(task_id):
    """Task that might fail"""
    await asyncio.sleep(0.5)
    
    if random.random() < 0.3:  # 30% chance of failure
        raise Exception(f"Task {task_id} failed!")
    
    return f"Task {task_id} succeeded"

async def robust_task_runner():
    """Run multiple tasks with proper error handling"""
    tasks = [unreliable_task(i) for i in range(5)]
    
    # Use return_exceptions to handle failures gracefully
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    successes = []
    failures = []
    
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            failures.append(f"Task {i}: {result}")
        else:
            successes.append(result)
    
    print(f"Successes: {len(successes)}")
    print(f"Failures: {len(failures)}")

asyncio.run(robust_task_runner())

import asyncio

async def number_generator(max_num):
    """Generate numbers asynchronously"""
    for i in range(max_num):
        print(f"Generating {i}")
        await asyncio.sleep(0.1)  # Simulate async work
        yield i

async def consume_numbers():
    """Consume numbers from async generator"""
    async for number in number_generator(5):
        print(f"Received: {number}")
        await asyncio.sleep(0.05)  # Process the number

asyncio.run(consume_numbers())

async def fetch_user_data(user_ids):
    """Simulate fetching user data from an API"""
    for user_id in user_ids:
        await asyncio.sleep(0.1)  # Simulate API call
        yield {
            "id": user_id,
            "name": f"User {user_id}",
            "email": f"user{user_id}@example.com"
        }

async def enrich_user_data(user_stream):
    """Add additional data to each user"""
    async for user in user_stream:
        await asyncio.sleep(0.05)  # Simulate enrichment
        
        user["profile_complete"] = len(user["name"]) > 5
        user["domain"] = user["email"].split("@")[1]
        yield user

async def process_users():
    """Complete user processing pipeline"""
    user_ids = [1, 2, 3, 4, 5]
    
    # Create the pipeline
    raw_users = fetch_user_data(user_ids)
    enriched_users = enrich_user_data(raw_users)
    
    # Process the pipeline
    saved_count = 0
    async for user in enriched_users:
        print(f"Saved: {user['name']} ({user['email']})")
        saved_count += 1
    
    print(f"Pipeline complete: {saved_count} users processed")

asyncio.run(process_users())

async def map_stream(source_stream, transform_func):
    """Apply transformation to each item in stream"""
    async for item in source_stream:
        yield transform_func(item)

async def filter_stream(source_stream, predicate):
    """Filter stream items based on predicate"""
    async for item in source_stream:
        if predicate(item):
            yield item

async def take_stream(source_stream, count):
    """Take only first N items from stream"""
    taken = 0
    async for item in source_stream:
        if taken >= count:
            break
        yield item
        taken += 1

async def number_stream():
    """Generate numbers"""
    for i in range(20):
        await asyncio.sleep(0.05)
        yield i

async def stream_operations():
    """Demonstrate stream transformations"""
    # Create pipeline: numbers -> squares -> evens -> first 5
    numbers = number_stream()
    squares = map_stream(numbers, lambda x: x * x)
    evens = filter_stream(squares, lambda x: x % 2 == 0)
    first_five = take_stream(evens, 5)
    
    results = []
    async for result in first_five:
        results.append(result)
        print(f"Result: {result}")
    
    print(f"Final results: {results}")

asyncio.run(stream_operations())

async def buffered_generator(source_stream, buffer_size=3):
    """Buffer items from source stream"""
    buffer = []
    
    async for item in source_stream:
        buffer.append(item)
        
        if len(buffer) >= buffer_size:
            yield buffer.copy()
            buffer.clear()
    
    # Yield remaining items
    if buffer:
        yield buffer

async def data_source():
    """Generate data items"""
    for i in range(10):
        await asyncio.sleep(0.1)
        yield f"item-{i}"

async def process_buffer(buffer):
    """Process a buffer of items"""
    print(f"Processing buffer of {len(buffer)} items: {buffer}")
    await asyncio.sleep(0.2)  # Simulate batch processing

async def buffered_processing():
    """Process data in buffers"""
    source = data_source()
    buffered = buffered_generator(source, buffer_size=3)
    
    async for buffer in buffered:
        await process_buffer(buffer)

asyncio.run(buffered_processing())

async def unreliable_data_source():
    """Data source that occasionally fails"""
    for i in range(10):
        await asyncio.sleep(0.1)
        
        if i == 5:  # Simulate error
            raise Exception(f"Data source error at item {i}")
        
        yield f"data-{i}"

async def resilient_stream(source_stream):
    """Make stream resilient to errors"""
    try:
        async for item in source_stream:
            yield item
    except Exception as e:
        print(f"Stream error: {e}")
        # Could implement retry logic here

async def error_handling_demo():
    """Demonstrate error handling in streams"""
    try:
        source = unreliable_data_source()
        resilient = resilient_stream(source)
        
        async for item in resilient:
            print(f"Processed: {item}")
            
    except Exception as e:
        print(f"Stream processing failed: {e}")

asyncio.run(error_handling_demo())

import asyncio
import random

async def risky_task(task_id):
    """Task that might fail"""
    await asyncio.sleep(0.5)
    
    if random.random() < 0.3:  # 30% failure rate
        raise ValueError(f"Task {task_id} failed!")
    
    return f"Task {task_id} completed"

async def basic_error_handling():
    """Basic error handling with try/except"""
    for i in range(5):
        try:
            result = await risky_task(i)
            print(f"Success: {result}")
        except ValueError as e:
            print(f"Error: {e}")

asyncio.run(basic_error_handling())

async def concurrent_error_handling():
    """Handle errors in concurrent tasks"""
    tasks = [risky_task(i) for i in range(10)]
    
    # Gather all results, including exceptions
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    successes = []
    failures = []
    
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            failures.append(f"Task {i}: {result}")
        else:
            successes.append(result)
    
    print(f"Successful tasks: {len(successes)}")
    print(f"Failed tasks: {len(failures)}")
    
    # Continue processing successful results
    for success in successes:
        print(f"Processing: {success}")

asyncio.run(concurrent_error_handling())

async def long_running_task(name, duration):
    """Task that can be cancelled"""
    try:
        print(f"{name}: Starting work for {duration}s")
        await asyncio.sleep(duration)
        print(f"{name}: Work completed")
        return f"{name} result"
    except asyncio.CancelledError:
        print(f"{name}: Task was cancelled, cleaning up...")
        # Perform cleanup here
        await asyncio.sleep(0.1)  # Simulate cleanup
        print(f"{name}: Cleanup complete")
        raise  # Re-raise to properly cancel

async def cancellation_demo():
    """Demonstrate task cancellation"""
    # Start multiple tasks
    tasks = [
        asyncio.create_task(long_running_task("Task-A", 3)),
        asyncio.create_task(long_running_task("Task-B", 5)),
        asyncio.create_task(long_running_task("Task-C", 2))
    ]
    
    try:
        # Wait for 2.5 seconds, then cancel remaining tasks
        await asyncio.sleep(2.5)
        
        print("🛑 Cancelling remaining tasks...")
        for task in tasks:
            if not task.done():
                task.cancel()
        
        # Wait for all tasks to finish (including cancellation)
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for i, result in enumerate(results):
            if isinstance(result, asyncio.CancelledError):
                print(f"Task {i}: Cancelled")
            elif isinstance(result, Exception):
                print(f"Task {i}: Failed - {result}")
            else:
                print(f"Task {i}: Completed - {result}")
                
    except KeyboardInterrupt:
        print("Received interrupt, cancelling all tasks...")
        for task in tasks:
            task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)

asyncio.run(cancellation_demo())

async def timeout_wrapper(coroutine, timeout_seconds, operation_name="Operation"):
    """Wrapper for timeout handling"""
    try:
        return await asyncio.wait_for(coroutine, timeout=timeout_seconds)
    except asyncio.TimeoutError:
        print(f"{operation_name} timed out after {timeout_seconds}s")
        raise
    except Exception as e:
        print(f"{operation_name} failed: {e}")
        raise

async def slow_operation():
    """Simulate slow operation"""
    await asyncio.sleep(3)
    return {"users": ["alice", "bob", "charlie"]}

async def timeout_demo():
    """Demonstrate timeout handling"""
    operations = [
        (slow_operation(), 2.5, "Database Query"),
        (slow_operation(), 1.5, "API Call"),
        (slow_operation(), 4.0, "File Processing")
    ]
    
    for coroutine, timeout, name in operations:
        try:
            result = await timeout_wrapper(coroutine, timeout, name)
            print(f"{name}: Success")
        except asyncio.TimeoutError:
            print(f"{name}: Timed out")
        except Exception as e:
            print(f"{name}: Failed - {e}")

asyncio.run(timeout_demo())

async def retry_with_backoff(
    coroutine_func, 
    max_retries=3, 
    base_delay=1, 
    backoff_factor=2
):
    """Retry with exponential backoff"""
    
    for attempt in range(max_retries + 1):
        try:
            return await coroutine_func()
        
        except Exception as e:
            if attempt == max_retries:
                print(f"🚫 All {max_retries + 1} attempts failed")
                raise e
            
            # Calculate delay with exponential backoff
            delay = base_delay * (backoff_factor ** attempt)
            
            print(f"Attempt {attempt + 1} failed: {e}")
            print(f"Retrying in {delay:.1f}s...")
            
            await asyncio.sleep(delay)

async def flaky_api_call():
    """API call that fails randomly"""
    await asyncio.sleep(0.2)
    
    if random.random() < 0.6:  # 60% failure rate
        raise Exception("API temporarily unavailable")
    
    return {"status": "success", "data": "API response"}

async def retry_demo():
    """Demonstrate retry with backoff"""
    try:
        result = await retry_with_backoff(
            flaky_api_call,
            max_retries=4,
            base_delay=0.5,
            backoff_factor=2
        )
        print(f"Success: {result}")
    
    except Exception as e:
        print(f"💥 Final failure: {e}")

asyncio.run(retry_demo())

import asyncio

async def waiter(event, name, work_duration):
    """Wait for event, then do work"""
    print(f"{name}: Waiting for event...")
    await event.wait()
    
    print(f"{name}: Event received, starting work")
    await asyncio.sleep(work_duration)
    print(f"{name}: Work completed")

async def event_setter(event, delay):
    """Set event after delay"""
    print(f"Preparing to set event in {delay}s...")
    await asyncio.sleep(delay)
    
    print("Setting event for all waiters")
    event.set()

async def event_demo():
    """Demonstrate event-based communication"""
    event = asyncio.Event()
    
    # Start multiple waiters and setter
    await asyncio.gather(
        waiter(event, "Worker-1", 1),
        waiter(event, "Worker-2", 2),
        event_setter(event, 2)
    )

asyncio.run(event_demo())

import random

async def producer(queue, items, producer_name):
    """Produce items with error handling"""
    try:
        for item in items:
            await queue.put(item)
            print(f"📦 {producer_name}: Produced {item}")
            await asyncio.sleep(0.1)
        
        # Signal completion
        await queue.put(None)
        print(f"🏁 {producer_name}: Production complete")
        
    except Exception as e:
        print(f"{producer_name}: Production failed - {e}")
        await queue.put(None)  # Still signal completion

async def consumer(queue, consumer_name):
    """Consume items with error handling"""
    processed = 0
    
    try:
        while True:
            try:
                item = await asyncio.wait_for(queue.get(), timeout=5.0)
            except asyncio.TimeoutError:
                print(f"{consumer_name}: Timeout waiting for items")
                break
            
            if item is None:  # End signal
                print(f"🏁 {consumer_name}: Received end signal")
                break
            
            await asyncio.sleep(0.2)  # Simulate processing
            processed += 1
            print(f"{consumer_name}: Processed {item}")
    
    except Exception as e:
        print(f"{consumer_name}: Consumer failed - {e}")
    
    finally:
        print(f"{consumer_name}: Processed {processed} items total")

async def producer_consumer_demo():
    """Robust producer-consumer with error handling"""
    queue = asyncio.Queue(maxsize=5)  # Bounded queue
    
    items = [f"item-{i}" for i in range(10)]
    
    # Start producer and consumers
    await asyncio.gather(
        producer(queue, items, "Producer-1"),
        consumer(queue, "Consumer-A"),
        consumer(queue, "Consumer-B"),
        return_exceptions=True  # Don't let one failure stop others
    )

asyncio.run(producer_consumer_demo())

async def rate_limited_operations(urls, max_concurrent=5):
    """Limit concurrent operations"""
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def fetch_with_limit(url):
        async with semaphore:
            return await fetch_url(url)
    
    tasks = [fetch_with_limit(url) for url in urls]
    return await asyncio.gather(*tasks)

async def fetch_url(url):
    await asyncio.sleep(0.5)  # Simulate network request
    return f"Data from {url}"

async def semaphore_demo():
    """Demonstrate semaphore usage"""
    urls = [f"https://api.example.com/data/{i}" for i in range(10)]
    
    print("Fetching URLs with concurrency limit...")
    results = await rate_limited_operations(urls, max_concurrent=3)
    
    print(f"Fetched {len(results)} URLs")

asyncio.run(semaphore_demo())

class AsyncCounter:
    def __init__(self):
        self.value = 0
        self.lock = asyncio.Lock()
    
    async def increment(self, worker_name):
        """Thread-safe increment"""
        async with self.lock:
            # Critical section
            current = self.value
            await asyncio.sleep(0.01)  # Simulate some work
            self.value = current + 1
            print(f"{worker_name}: Incremented to {self.value}")

async def worker(counter, worker_name, iterations):
    """Worker that increments counter"""
    for i in range(iterations):
        await counter.increment(f"{worker_name}-{i}")
        await asyncio.sleep(0.05)

async def lock_demo():
    """Demonstrate async lock usage"""
    counter = AsyncCounter()
    
    # Start multiple workers
    await asyncio.gather(
        worker(counter, "Worker-A", 5),
        worker(counter, "Worker-B", 5)
    )
    
    print(f"Final counter value: {counter.value}")

asyncio.run(lock_demo())

async def worker_with_barrier(barrier, worker_name, work_time):
    """Worker that synchronizes at barrier"""
    
    # Phase 1: Individual work
    print(f"{worker_name}: Starting phase 1")
    await asyncio.sleep(work_time)
    print(f"{worker_name}: Phase 1 complete")
    
    # Synchronization point
    print(f"{worker_name}: Waiting at barrier")
    await barrier.wait()
    
    # Phase 2: Coordinated work
    print(f"{worker_name}: Starting phase 2")
    await asyncio.sleep(0.5)
    print(f"{worker_name}: Phase 2 complete")

async def barrier_demo():
    """Demonstrate barrier synchronization"""
    # Create barrier for 3 workers
    barrier = asyncio.Barrier(3)
    
    # Start workers with different work times
    await asyncio.gather(
        worker_with_barrier(barrier, "Worker-A", 1.0),
        worker_with_barrier(barrier, "Worker-B", 2.0),
        worker_with_barrier(barrier, "Worker-C", 1.5)
    )

asyncio.run(barrier_demo())

from fastapi import FastAPI, HTTPException
import asyncio

app = FastAPI(title="My Async API")

@app.get("/")
async def root():
    return {"message": "Hello Async World!"}

@app.get("/slow-endpoint")
async def slow_endpoint():
    await asyncio.sleep(2)  # Simulate slow database query
    return {"message": "This took 2 seconds, but didn't block other requests!"}

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # Simulate async database call
    await asyncio.sleep(0.1)
    
    if user_id == 404:
        raise HTTPException(status_code=404, detail="User not found")
    
    return {
        "id": user_id,
        "name": f"User {user_id}",
        "email": f"user{user_id}@example.com"
    }

# Run with: uvicorn main:app --reload

import asyncpg
from fastapi import Depends

# Database connection pool
db_pool = None

@app.on_event("startup")
async def startup():
    global db_pool
    db_pool = await asyncpg.create_pool(
        "postgresql://user:password@localhost/mydb",
        min_size=5,
        max_size=20
    )

@app.on_event("shutdown")
async def shutdown():
    if db_pool:
        await db_pool.close()

async def get_db():
    """Dependency to get database connection"""
    async with db_pool.acquire() as connection:
        yield connection

@app.get("/users")
async def list_users(limit: int = 10, db=Depends(get_db)):
    """List users with pagination"""
    query = "SELECT id, name, email FROM users LIMIT $1"
    rows = await db.fetch(query, limit)
    
    return [
        {"id": row["id"], "name": row["name"], "email": row["email"]}
        for row in rows
    ]

@app.post("/users")
async def create_user(user_data: dict, db=Depends(get_db)):
    """Create a new user"""
    query = """
        INSERT INTO users (name, email) 
        VALUES ($1, $2) 
        RETURNING id, name, email
    """
    
    row = await db.fetchrow(query, user_data["name"], user_data["email"])
    
    return {"id": row["id"], "name": row["name"], "email": row["email"]}

import aiohttp

async def fetch_weather(city: str) -> dict:
    """Fetch weather data from external API"""
    async with aiohttp.ClientSession() as session:
        url = f"https://api.weather.com/v1/current?city={city}"
        
        try:
            async with session.get(url, timeout=5) as response:
                if response.status == 200:
                    return await response.json()
                else:
                    return {"error": "Weather service unavailable"}
        except asyncio.TimeoutError:
            return {"error": "Weather service timeout"}

async def fetch_news(category: str) -> dict:
    """Fetch news from external API"""
    async with aiohttp.ClientSession() as session:
        url = f"https://api.news.com/v1/headlines?category={category}"
        
        try:
            async with session.get(url, timeout=5) as response:
                if response.status == 200:
                    return await response.json()
                else:
                    return {"articles": []}  # Graceful degradation
        except:
            return {"articles": []}  # Graceful degradation

@app.get("/dashboard/{city}")
async def get_dashboard(city: str):
    """Get dashboard data from multiple sources concurrently"""
    
    # Fetch data from multiple sources simultaneously
    weather, news = await asyncio.gather(
        fetch_weather(city),
        fetch_news("technology"),
        return_exceptions=True
    )
    
    # Handle partial failures gracefully
    dashboard = {"city": city}
    
    if isinstance(weather, dict):
        dashboard["weather"] = weather
    else:
        dashboard["weather"] = {"error": "Weather unavailable"}
    
    if isinstance(news, dict):
        dashboard["news"] = news
    else:
        dashboard["news"] = {"articles": []}
    
    return dashboard

from fastapi import BackgroundTasks
import logging

logger = logging.getLogger(__name__)

async def send_email(email: str, subject: str):
    """Simulate sending email"""
    logger.info(f"Sending email to {email}: {subject}")
    await asyncio.sleep(2)  # Simulate email sending
    logger.info(f"Email sent to {email}")

async def process_image(image_path: str, user_id: int):
    """Simulate image processing"""
    logger.info(f"Processing image {image_path} for user {user_id}")
    await asyncio.sleep(5)  # Simulate processing
    logger.info(f"Image processing complete for user {user_id}")

@app.post("/users/{user_id}/upload")
async def upload_image(
    user_id: int,
    image_data: dict,
    background_tasks: BackgroundTasks
):
    """Upload and process image"""
    
    # Save image immediately
    image_path = f"/uploads/{user_id}/{image_data['filename']}"
    
    # Process image in background
    background_tasks.add_task(process_image, image_path, user_id)
    
    # Send confirmation email in background
    background_tasks.add_task(
        send_email,
        "[email protected]",
        "Image Upload Confirmation"
    )
    
    return {
        "message": "Image uploaded successfully",
        "image_path": image_path,
        "status": "processing"
    }

import time
from collections import defaultdict
from fastapi import Request, HTTPException

class RateLimiter:
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = defaultdict(list)
    
    async def is_allowed(self, client_id: str) -> bool:
        """Check if request is allowed"""
        now = time.time()
        
        # Clean old requests
        self.requests[client_id] = [
            req_time for req_time in self.requests[client_id]
            if now - req_time < self.window_seconds
        ]
        
        # Check if under limit
        if len(self.requests[client_id]) < self.max_requests:
            self.requests[client_id].append(now)
            return True
        
        return False

# Global rate limiter: 100 requests per minute
rate_limiter = RateLimiter(max_requests=100, window_seconds=60)

async def rate_limit_middleware(request: Request, call_next):
    """Rate limiting middleware"""
    client_ip = request.client.host
    
    if not await rate_limiter.is_allowed(client_ip):
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded. Try again later."
        )
    
    response = await call_next(request)
    return response

app.middleware("http")(rate_limit_middleware)

import asyncio
import aiofiles
from pathlib import Path
import json

async def process_single_file(file_path: Path):
    """Process a single JSON file"""
    try:
        async with aiofiles.open(file_path, 'r') as file:
            content = await file.read()
            data = json.loads(content)
        
        record_count = len(data) if isinstance(data, list) else 1
        
        return {
            "file": file_path.name,
            "records": record_count,
            "status": "success"
        }
    
    except Exception as e:
        return {
            "file": file_path.name,
            "error": str(e),
            "status": "failed"
        }

async def process_files_batch(file_paths, batch_size=10):
    """Process files in batches to control memory usage"""
    results = []
    
    for i in range(0, len(file_paths), batch_size):
        batch = file_paths[i:i + batch_size]
        
        print(f"Processing batch {i//batch_size + 1}: {len(batch)} files")
        
        # Process batch concurrently
        batch_tasks = [process_single_file(path) for path in batch]
        batch_results = await asyncio.gather(*batch_tasks)
        
        results.extend(batch_results)
        
        # Optional: brief pause between batches
        await asyncio.sleep(0.1)
    
    return results

async def file_processing_demo():
    """Demonstrate concurrent file processing"""
    # Simulate file paths
    file_paths = [Path(f"data/file_{i}.json") for i in range(50)]
    
    start_time = asyncio.get_event_loop().time()
    results = await process_files_batch(file_paths, batch_size=10)
    elapsed = asyncio.get_event_loop().time() - start_time
    
    successful = len([r for r in results if r["status"] == "success"])
    failed = len([r for r in results if r["status"] == "failed"])
    
    print(f"Processed {len(results)} files in {elapsed:.2f}s")
    print(f"Successful: {successful}")
    print(f"Failed: {failed}")

asyncio.run(file_processing_demo())

import aiohttp

async def extract_api_data(api_urls):
    """Extract data from multiple APIs"""
    async with aiohttp.ClientSession() as session:
        for url in api_urls:
            try:
                async with session.get(url, timeout=10) as response:
                    if response.status == 200:
                        data = await response.json()
                        yield data
                    else:
                        print(f"Failed to fetch {url}: {response.status}")
            except Exception as e:
                print(f"Error fetching {url}: {e}")

async def transform_data(data_stream):
    """Transform raw data"""
    async for raw_data in data_stream:
        # Clean and transform data
        transformed = {
            "id": raw_data.get("id"),
            "name": raw_data.get("name", "").strip().title(),
            "email": raw_data.get("email", "").lower(),
            "processed_at": asyncio.get_event_loop().time()
        }
        
        # Skip invalid records
        if transformed["id"] and transformed["email"]:
            yield transformed

async def load_to_database(transformed_stream):
    """Load transformed data to database (simulated)"""
    batch = []
    batch_size = 100
    
    async for record in transformed_stream:
        batch.append(record)
        
        if len(batch) >= batch_size:
            await insert_batch(batch)
            batch.clear()
    
    # Insert remaining records
    if batch:
        await insert_batch(batch)

async def insert_batch(records):
    """Insert batch of records to database (simulated)"""
    # Simulate database insert
    await asyncio.sleep(0.1)
    print(f"Inserted batch of {len(records)} records")

    # Load to database
    await load_to_database(transformed_data)
    
    print("ETL pipeline completed successfully")

asyncio.run(etl_pipeline())

import random
from datetime import datetime

async def data_stream_simulator():
    """Simulate real-time data stream"""
    while True:
        # Generate sample event
        event = {
            "timestamp": datetime.utcnow().isoformat(),
            "user_id": random.randint(1, 1000),
            "event_type": random.choice(["login", "logout", "purchase", "view"]),
            "value": random.uniform(0, 100)
        }
        
        yield event
        await asyncio.sleep(0.1)  # 10 events per second

async def event_aggregator(event_stream, window_seconds=5):
    """Aggregate events in time windows"""
    window_data = {}
    last_flush = asyncio.get_event_loop().time()
    
    async for event in event_stream:
        current_time = asyncio.get_event_loop().time()
        
        # Add event to current window
        event_type = event["event_type"]
        if event_type not in window_data:
            window_data[event_type] = {"count": 0, "total_value": 0}
        
        window_data[event_type]["count"] += 1
        window_data[event_type]["total_value"] += event["value"]
        
        # Flush window if time elapsed
        if current_time - last_flush >= window_seconds:
            if window_data:
                yield {
                    "window_start": last_flush,
                    "window_end": current_time,
                    "aggregates": window_data.copy()
                }
                
                window_data.clear()
                last_flush = current_time

async def process_aggregates(aggregate_stream):
    """Process aggregated data"""
    async for window in aggregate_stream:
        print(f"Window {window['window_start']:.1f}-{window['window_end']:.1f}:")
        for event_type, stats in window["aggregates"].items():
            avg_value = stats["total_value"] / stats["count"]
            print(f"  {event_type}: {stats['count']} events, avg value: {avg_value:.1f}")

async def real_time_processing():
    """Real-time data processing pipeline"""
    # Create processing pipeline
    events = data_stream_simulator()
    aggregates = event_aggregator(events, window_seconds=3)
    
    # Process aggregates (this will run indefinitely)
    try:
        await process_aggregates(aggregates)
    except KeyboardInterrupt:
        print("Stopping real-time processing...")

# Run for a limited time in demo
async def demo_real_time():
    task = asyncio.create_task(real_time_processing())
    
    # Run for 15 seconds
    await asyncio.sleep(15)
    task.cancel()
    
    try:
        await task
    except asyncio.CancelledError:
        print("Demo completed")

asyncio.run(demo_real_time())

import asyncio

async def enrich_user_data(user_id: int):
    """Enrich user data from multiple sources"""
    
    async def get_profile(user_id):
        await asyncio.sleep(0.2)  # Simulate API call
        return {"name": f"User {user_id}", "age": 25 + (user_id % 40)}
    
    async def get_preferences(user_id):
        await asyncio.sleep(0.3)  # Simulate API call
        return {"theme": "dark", "notifications": True}
    
    async def get_activity(user_id):
        await asyncio.sleep(0.1)  # Simulate API call
        return {"last_login": "2024-01-01", "login_count": user_id * 10}

    # Fetch all data concurrently
    profile, preferences, activity = await asyncio.gather(
        get_profile(user_id),
        get_preferences(user_id),
        get_activity(user_id),
        return_exceptions=True
    )

    # Combine results, handling failures gracefully
    enriched_data = {"user_id": user_id}
    
    if isinstance(profile, dict):
        enriched_data.update(profile)
    
    if isinstance(preferences, dict):
        enriched_data["preferences"] = preferences
    
    if isinstance(activity, dict):
        enriched_data["activity"] = activity
    
    return enriched_data

async def batch_enrich_users(user_ids, batch_size=20):
    """Enrich multiple users in batches"""
    results = []
    
    for i in range(0, len(user_ids), batch_size):
        batch = user_ids[i:i + batch_size]
        
        print(f"Enriching batch {i//batch_size + 1}: {len(batch)} users")
        
        # Enrich batch concurrently
        batch_tasks = [enrich_user_data(uid) for uid in batch]
        batch_results = await asyncio.gather(*batch_tasks)
        
        results.extend(batch_results)
    
    return results

async def enrichment_demo():
    """Demonstrate parallel data enrichment"""
    user_ids = list(range(1, 51))  # 50 users
    
    start_time = asyncio.get_event_loop().time()
    enriched_users = await batch_enrich_users(user_ids, batch_size=10)
    elapsed = asyncio.get_event_loop().time() - start_time
    
    print(f"Enriched {len(enriched_users)} users in {elapsed:.2f}s")

asyncio.run(enrichment_demo())

import re

async def validate_email(email: str) -> bool:
    """Validate email format"""
    await asyncio.sleep(0.01)  # Simulate validation work
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return bool(re.match(pattern, email))

async def validate_phone(phone: str) -> bool:
    """Validate phone number"""
    await asyncio.sleep(0.01)  # Simulate validation work
    cleaned = re.sub(r'[^\d]', '', phone)
    return 10 <= len(cleaned) <= 15

async def validate_age(age) -> bool:
    """Validate age"""
    await asyncio.sleep(0.005)  # Simulate validation work
    try:
        age_int = int(age)
        return 0 <= age_int <= 150
    except (ValueError, TypeError):
        return False

async def validate_record(record: dict) -> dict:
    """Validate a single record"""
    # Run all validations concurrently
    email_valid, phone_valid, age_valid = await asyncio.gather(
        validate_email(record.get("email", "")),
        validate_phone(record.get("phone", "")),
        validate_age(record.get("age")),
        return_exceptions=True
    )
    
    validation_results = {
        "email_valid": email_valid if isinstance(email_valid, bool) else False,
        "phone_valid": phone_valid if isinstance(phone_valid, bool) else False,
        "age_valid": age_valid if isinstance(age_valid, bool) else False
    }
    
    # Overall validity
    validation_results["is_valid"] = all(validation_results.values())
    
    return {
        **record,
        "validation": validation_results
    }

async def validation_pipeline(records):
    """Validate multiple records"""
    batch_size = 50
    validated_records = []
    
    for i in range(0, len(records), batch_size):
        batch = records[i:i + batch_size]
        
        # Validate batch concurrently
        batch_tasks = [validate_record(record) for record in batch]
        batch_results = await asyncio.gather(*batch_tasks)
        
        validated_records.extend(batch_results)
        
        print(f"Validated batch {i//batch_size + 1}: {len(batch)} records")
    
    return validated_records

async def validation_demo():
    """Demonstrate data validation pipeline"""
    # Sample data with various quality issues
    sample_records = [
        {"name": "John Doe", "email": "[email protected]", "phone": "555-1234", "age": 30},
        {"name": "Jane Smith", "email": "invalid-email", "phone": "555-5678", "age": 25},
        {"name": "Bob Johnson", "email": "[email protected]", "phone": "not-a-phone", "age": "thirty"},
    ] * 50  # Multiply to simulate larger dataset
    
    start_time = asyncio.get_event_loop().time()
    validated = await validation_pipeline(sample_records)
    elapsed = asyncio.get_event_loop().time() - start_time
    
    # Analyze results
    valid_count = len([r for r in validated if r["validation"]["is_valid"]])
    invalid_count = len(validated) - valid_count
    
    print(f"\nValidation Results:")
    print(f"Total records: {len(validated)}")
    print(f"Valid: {valid_count}")
    print(f"Invalid: {invalid_count}")
    print(f"Processing time: {elapsed:.2f}s")

asyncio.run(validation_demo())