pythonintermediate

asyncio.gather Concurrent Tasks

Run multiple async operations concurrently with asyncio.gather and proper error handling.

pythonPress ⌘/Ctrl + Shift + C to copy

import asyncio
from typing import Any

async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.5)  # Simulate API call
    return {"id": user_id, "name": f"User {user_id}"}

async def fetch_orders(user_id: int) -> list:
    await asyncio.sleep(0.3)
    return [{"id": 1, "total": 99.99}]

async def fetch_profile(user_id: int) -> dict[str, Any]:
    # Run all three concurrently
    user, orders, settings = await asyncio.gather(
        fetch_user(user_id),
        fetch_orders(user_id),
        fetch_user(user_id),  # placeholder for settings
        return_exceptions=True,  # Don't fail all on one error
    )

    # Handle individual failures
    if isinstance(user, Exception):
        user = {"error": str(user)}

    return {"user": user, "orders": orders, "settings": settings}

# asyncio.run(fetch_profile(42))