</>
pythonintermediate

asyncio.gather Concurrent Tasks

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

python
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))

Use Cases

  • Parallel API calls
  • Concurrent database queries
  • Batch processing

Tags

#asyncio#concurrency#async#parallel

Related Snippets

Similar patterns you can reuse in the same workflow.

pythonintermediate

Async ETL Pipeline with asyncio

Run concurrent data fetches and transformations using asyncio.gather for high-throughput pipelines.

Best for: concurrent API ingestion

#asyncio#async
pythonintermediate

Python Concurrent Futures for Parallel Work

Run tasks in parallel using ThreadPoolExecutor and ProcessPoolExecutor with error handling.

Best for: Parallel HTTP requests for web scraping

#python#concurrency
pythonadvanced

Run Async Tasks Concurrently with gather

Execute multiple async operations concurrently using asyncio.gather.

Best for: Parallel API calls

#python#asyncio
pythonadvanced

Asyncio Semaphore and Timeout Patterns

Control concurrency with asyncio semaphores, timeouts, and task groups for robust async code.

Best for: rate limiting API calls

#python#asyncio