Concurrent I/O with asyncio

import asyncio

# Define an async function (coroutine)
async def fetch_status(url: str) -> dict:
    """Fetch status from a service endpoint."""
    print(f"Fetching {url}...")
    await asyncio.sleep(1)  # Simulates I/O (network call)
    return {"url": url, "status": "healthy"}

# Run a single coroutine
async def main():
    result = await fetch_status("http://api.example.com/health")
    print(result)

asyncio.run(main())

import asyncio
import aiohttp

async def check_service(session: aiohttp.ClientSession, url: str) -> dict:
    """Check a single service endpoint."""
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
            return {"url": url, "status": resp.status, "ok": resp.status == 200}
    except Exception as e:
        return {"url": url, "status": 0, "ok": False, "error": str(e)}

async def check_all_services(urls: list[str]) -> list[dict]:
    """Check all services concurrently."""
    async with aiohttp.ClientSession() as session:
        tasks = [check_service(session, url) for url in urls]
        return await asyncio.gather(*tasks)

# Usage
urls = [
    "http://web-01:8080/health",
    "http://web-02:8080/health",
    "http://api-01:8080/health",
    "http://db-01:5432/health",
]

results = asyncio.run(check_all_services(urls))
for r in results:
    status = "UP" if r["ok"] else "DOWN"
    print(f"{status}  {r['url']}")

import asyncio
import aiohttp

async def fetch_with_retry(session: aiohttp.ClientSession, url: str, retries: int = 3) -> dict:
    for attempt in range(retries):
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
                return {"url": url, "status": resp.status}
        except Exception:
            if attempt == retries - 1:
                return {"url": url, "status": 0, "error": "max retries"}
            await asyncio.sleep(0.5 * (attempt + 1))  # Exponential backoff

async def main():
    urls = ["http://web-01:8080/health", "http://web-02:8080/health"]

    async with aiohttp.ClientSession() as session:
        # TaskGroup ensures all tasks complete or all cancel on error
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch_with_retry(session, url)) for url in urls]

        # All tasks are done here
        for task in tasks:
            print(task.result())

asyncio.run(main())

from concurrent.futures import ThreadPoolExecutor, as_completed
import subprocess

def run_command(cmd: str) -> dict:
    """Run a shell command in a thread."""
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=30)
    return {"cmd": cmd, "stdout": result.stdout.strip(), "exit": result.returncode}

# Run multiple commands concurrently
commands = ["uptime", "df -h", "free -m", "ps aux | wc -l"]

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(run_command, cmd): cmd for cmd in commands}
    for future in as_completed(futures):
        result = future.result()
        print(f"{result['cmd']}: exit {result['exit']}")
        print(f"  {result['stdout']}")

from concurrent.futures import ProcessPoolExecutor
import hashlib
from pathlib import Path

def hash_file(path: str) -> dict:
    """Compute SHA256 hash of a file."""
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(8192), b""):
            h.update(chunk)
    return {"path": path, "hash": h.hexdigest()}

# Hash files across multiple CPU cores
files = [str(p) for p in Path("/var/log").glob("**/*.log")]

with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(hash_file, files))

for r in results:
    print(f"{r['hash'][:12]}  {r['path']}")