Python Concurrency — asyncio Deep Dive for Senior Engineers

import asyncio
import time

# A production-grade coroutine with proper type hinting
async def fetch_service_status(service_name: str, delay: float) -> str:
    print(f"[io.thecodeforge] Requesting status for {service_name}...")
    # await suspends this coroutine, allowing the loop to process other tasks
    await asyncio.sleep(delay) 
    return f"{service_name}: UP"

async def main():
    start_time = time.perf_counter()

    # SEQUENTIAL PATTERN: Anti-pattern for independent tasks
    status_a = await fetch_service_status("Auth-Service", 1.0)
    status_b = await fetch_service_status("Payment-Gateway", 1.0)
    
    total_time = time.perf_counter() - start_time
    print(f"Total Sequential Time: {total_time:.2f}s") # Results in ~2.00s

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import time

async def main():
    start_time = time.perf_counter()

    # CONCURRENT PATTERN: Launching tasks in parallel
    # gather() wraps coroutines into Tasks and waits for all results
    results = await asyncio.gather(
        fetch_service_status("Database", 1.5),
        fetch_service_status("Cache", 0.5),
        fetch_service_status("Search-Index", 1.2)
    )

    total_time = time.perf_counter() - start_time
    print(f"Concurrent Results: {results}")
    print(f"Total Concurrent Time: {total_time:.2f}s") # Results in ~1.50s

# Reusing the fetch_service_status from above
asyncio.run(main())

import asyncio

async def api_call(name: str, should_fail: bool = False):
    await asyncio.sleep(0.2)
    if should_fail:
        raise RuntimeError(f"Critical failure in {name}")
    return f"{name}_data"

async def main():
    # Defensive Gathering
    tasks = [api_call("API_1"), api_call("API_2", True), api_call("API_3")]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    for res in results:
        if isinstance(res, Exception):
            print(f"Managed Error: {res}")
        else:
            print(f"Fetched: {res}")

    # Enforcing Strict Timeouts
    try:
        await asyncio.wait_for(api_call("Slow_API"), timeout=0.1)
    except asyncio.TimeoutError:
        print("Slow_API exceeded SLA and was cancelled.")

asyncio.run(main())

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

def heavy_cpu_bound_task(data):
    # This is synchronous and would normally block the loop
    time.sleep(2) 
    return f"Processed {data}"

async def main():
    loop = asyncio.get_running_loop()
    
    # Offload blocking work to a separate thread pool
    # This keeps the main Event Loop free to handle other I/O
    with ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, heavy_cpu_bound_task, "BigData")
        print(result)

    # Recommended async library: httpx (replaces 'requests')
    # import httpx
    # async with httpx.AsyncClient() as client:
    #     resp = await client.get('https://thecodeforge.io')

asyncio.run(main())