Python Async Programming: asyncio, Concurrency Patterns, and Performance Tips

import asyncio
import time

# ❌ Blocking (synchronous) code
def fetch_data_sync(n):
    """Simulates API call"""
    time.sleep(2)  # Blocks entire program
    return f"Data {n}"

def main_sync():
    start = time.time()
    results = []
    for i in range(5):
        results.append(fetch_data_sync(i))
    print(f"Total time: {time.time() - start:.2f}s")  # ~10 seconds

# ✅ Non-blocking (asynchronous) code
async def fetch_data_async(n):
    """Async API call"""
    await asyncio.sleep(2)  # Yields control to event loop
    return f"Data {n}"

async def main_async():
    start = time.time()
    # Run all coroutines concurrently
    results = await asyncio.gather(
        fetch_data_async(0),
        fetch_data_async(1),
        fetch_data_async(2),
        fetch_data_async(3),
        fetch_data_async(4)
    )
    print(f"Total time: {time.time() - start:.2f}s")  # ~2 seconds

# Run async function
asyncio.run(main_async())

# Key concepts:
# - Coroutine: async function that can be paused/resumed
# - await: Pause execution until coroutine completes
# - Event loop: Manages execution of coroutines
# - asyncio.gather(): Run multiple coroutines concurrently

import aiohttp
import asyncio

# 1. Concurrent HTTP requests
async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def fetch_multiple_urls(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

# 2. Async context managers
class AsyncDatabase:
    async def __aenter__(self):
        self.conn = await asyncpg.connect('postgresql://localhost')
        return self.conn
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.conn.close()

async def query_database():
    async with AsyncDatabase() as conn:
        result = await conn.fetch('SELECT * FROM users')
        return result

# 3. Async generators for streaming
async def fetch_logs_stream(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            async for line in response.content:
                yield line.decode('utf-8')

async def process_logs():
    async for log in fetch_logs_stream('http://api.example.com/logs'):
        print(log)
        if should_stop(log):
            break

# 4. Async task queue with semaphore
async def worker(semaphore, queue, results):
    async with semaphore:  # Limit concurrent workers
        while True:
            url = await queue.get()
            if url is None:
                break
            result = await fetch_url(url)
            results.append(result)
            queue.task_done()

async def process_urls_with_limit(urls, max_concurrent=10):
    semaphore = asyncio.Semaphore(max_concurrent)
    queue = asyncio.Queue()
    results = []
    
    # Add URLs to queue
    for url in urls:
        await queue.put(url)
    
    # Create workers
    workers = [
        asyncio.create_task(worker(semaphore, queue, results))
        for _ in range(max_concurrent)
    ]
    
    await queue.join()  # Wait for all tasks
    
    # Stop workers
    for _ in range(max_concurrent):
        await queue.put(None)
    
    await asyncio.gather(*workers)
    return results

import asyncio
import threading
import multiprocessing
import time

# Decision Matrix:

# 1. I/O-Bound Tasks (API calls, DB queries, file I/O)
# ✅ Use asyncio
async def io_bound_async():
    tasks = [asyncio.sleep(1) for _ in range(100)]
    await asyncio.gather(*tasks)
    # Completes in ~1 second, single thread

# 2. I/O-Bound with blocking libraries
# ✅ Use threading (when library doesn't support async)
def io_bound_threading():
    import requests
    def fetch(url):
        return requests.get(url).text
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        urls = ['http://example.com'] * 100
        results = list(executor.map(fetch, urls))

# 3. CPU-Bound Tasks (data processing, calculations)
# ✅ Use multiprocessing
def cpu_bound_task(n):
    return sum(i * i for i in range(n))

def cpu_bound_multiprocessing():
    with multiprocessing.Pool() as pool:
        results = pool.map(cpu_bound_task, [10000000] * 4)
    # Uses multiple CPU cores

# ❌ Don't use asyncio for CPU-bound (GIL prevents parallelism)
async def cpu_bound_async_bad():  # This won't help!
    await asyncio.gather(*[
        asyncio.to_thread(cpu_bound_task, 10000000)
        for _ in range(4)
    ])

# Summary:
# asyncio: I/O-bound, many connections, single core
# threading: I/O-bound with blocking libs, shared state
# multiprocessing: CPU-bound, need parallelism, no GIL

import asyncio

# 1. Timeouts
async def fetch_with_timeout(url, timeout=5):
    try:
        async with asyncio.timeout(timeout):
            # Operation must complete within 5 seconds
            result = await fetch_data(url)
            return result
    except asyncio.TimeoutError:
        print(f"Request to {url} timed out")
        return None

# 2. Handle individual task failures
async def fetch_all_ignore_errors(urls):
    tasks = [fetch_with_timeout(url) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # Filter out errors
    successful = [r for r in results if not isinstance(r, Exception)]
    failed = [r for r in results if isinstance(r, Exception)]
    
    return successful, failed

# 3. Retry with exponential backoff
async def fetch_with_retry(url, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await fetch_data(url)
        except aiohttp.ClientError as e:
            if attempt == max_retries - 1:
                raise
            wait_time = 2 ** attempt  # 1s, 2s, 4s
            print(f"Retry {attempt + 1} after {wait_time}s")
            await asyncio.sleep(wait_time)

# 4. Graceful shutdown
async def main():
    tasks = []
    try:
        # Create long-running tasks
        tasks = [
            asyncio.create_task(worker(i))
            for i in range(10)
        ]
        await asyncio.gather(*tasks)
    except KeyboardInterrupt:
        print("Shutting down gracefully...")
        # Cancel all tasks
        for task in tasks:
            task.cancel()
        # Wait for cancellation
        await asyncio.gather(*tasks, return_exceptions=True)

# 5. Task groups (Python 3.11+)
async def main_with_taskgroup():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_data(1))
        task2 = tg.create_task(fetch_data(2))
    # If any task fails, all tasks are cancelled