From Blocking Loop to Async Pipeline: Parallelizing API Calls with asyncio

import asyncio
import os
import logging
from dataclasses import dataclass

import httpx

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

# Configuration from environment — never hardcode secrets or endpoints
API_URL = os.environ["PRODUCT_API_URL"]
API_KEY = os.environ["PRODUCT_API_KEY"]

# Concurrency controls — protect the remote service and your own system
MAX_CONCURRENCY = 20       # At most 20 requests in flight at once
REQUEST_TIMEOUT = 10.0     # Per-request timeout in seconds
MAX_RETRIES = 3            # Retry transient failures up to 3 times
BACKOFF_BASE = 0.5         # Base delay for exponential backoff


@dataclass
class FetchResult:
  """Encapsulates the outcome of a single API call."""
  product_id: int
  data: dict | None = None
  error: str | None = None
  success: bool = True


async def fetch_product(
  client: httpx.AsyncClient,
  semaphore: asyncio.Semaphore,
  product_id: int,
) -> FetchResult:
  """Fetch a single product with bounded concurrency and retry logic."""
  async with semaphore:
      # Semaphore limits how many coroutines enter this block simultaneously
      for attempt in range(MAX_RETRIES):
          try:
              response = await client.get(f"{API_URL}/products/{product_id}")
              response.raise_for_status()
              return FetchResult(product_id=product_id, data=response.json())

          except httpx.HTTPStatusError as e:
              if e.response.status_code == 429:
                  # Rate limited — back off and retry
                  wait = float(e.response.headers.get(
                      "Retry-After", BACKOFF_BASE * (2 ** attempt)
                  ))
                  logger.warning(
                      f"Rate limited on product {product_id}. "
                      f"Waiting {wait}s (attempt {attempt + 1}/{MAX_RETRIES})"
                  )
                  await asyncio.sleep(wait)
              elif e.response.status_code >= 500:
                  # Server error — retry with backoff
                  wait = BACKOFF_BASE * (2 ** attempt)
                  logger.warning(
                      f"Server error {e.response.status_code} for product {product_id}. "
                      f"Retrying in {wait}s"
                  )
                  await asyncio.sleep(wait)
              else:
                  # Client error (4xx) — do not retry, this is our fault
                  logger.error(f"Client error for product {product_id}: {e.response.status_code}")
                  return FetchResult(
                      product_id=product_id, error=str(e), success=False
                  )

          except httpx.TransportError as e:
              # Network failure — retry with backoff
              wait = BACKOFF_BASE * (2 ** attempt)
              logger.warning(
                  f"Transport error for product {product_id}: {e}. "
                  f"Retrying in {wait}s"
              )
              await asyncio.sleep(wait)

      # All retries exhausted for this product
      logger.error(f"All {MAX_RETRIES} attempts failed for product {product_id}")
      return FetchResult(
          product_id=product_id,
          error=f"Failed after {MAX_RETRIES} retries",
          success=False,
      )


async def fetch_all_products(product_ids: list[int]) -> list[FetchResult]:
  """Fetch all products concurrently with bounded parallelism."""
  semaphore = asyncio.Semaphore(MAX_CONCURRENCY)

  async with httpx.AsyncClient(
      headers={"Authorization": f"Bearer {API_KEY}"},
      timeout=httpx.Timeout(REQUEST_TIMEOUT, connect=5.0),
  ) as client:
      # Launch all coroutines at once — the semaphore gates actual execution
      tasks = [
          fetch_product(client, semaphore, pid)
          for pid in product_ids
      ]
      results = await asyncio.gather(*tasks)

  succeeded = sum(1 for r in results if r.success)
  failed = sum(1 for r in results if not r.success)
  logger.info(f"Completed: {succeeded} succeeded, {failed} failed out of {len(results)} total")
  return list(results)


if __name__ == "__main__":
  import time
  product_ids = list(range(1, 501))
  start = time.perf_counter()
  results = asyncio.run(fetch_all_products(product_ids))
  elapsed = time.perf_counter() - start
  logger.info(f"Total time: {elapsed:.2f}s")