Async Python Cloud APIs

If you've ever written a script that loops through a list of cloud instances, buckets, or Kubernetes clusters and found yourself waiting seconds — sometimes minutes — for each API call to finish, you know the pain. When your DevOps automation grows from a handful of resources to hundreds or thousands, synchronous requests become a bottleneck that turns a simple inventory check into a coffee-break operation. This lesson shows you how to query cloud APIs with async Python — the same pattern used by modern tooling — to cut those wait times dramatically, without adding complexity to your everyday automation scripts.

The Problem: Your Scripts Are Slow Because You Wait

Every time your Python script makes a cloud API call — whether it's boto3.list_instances(), azure.mgmt.compute() or kubectl.get_pods() — it sends a request, then blocks your entire program while it waits for the response. This is fine for one or two calls, but automation rarely stops there.

Consider a typical task: fetch status of every EC2 instance across all regions and print a summary. That's 20+ regions × 4 API calls each. If each call takes 300 ms (network + cloud service processing), you're staring at a blank screen for 24 seconds. Multiply that by every time you run it — daily, hourly, or as part of a CI/CD pipeline — and you're losing real time.

The core problem: synchronous I/O means your program spends most of its time waiting, not computing. For DevOps automation, where you often need to gather data from many sources before making decisions, this is pure waste.

Core Concept / Mental Model

Think of your script as a restaurant kitchen. In synchronous cooking, one chef (your program thread) prepares dish A, waits for it to bake, then starts dish B. But if you hire more chefs (threads), they can each work on a dish — but they'd step on each other in the same kitchen, and coordinating them is hard.

Async Python is more like a chef who starts dish A, then while it's in the oven, starts prepping dish B, then C, then comes back to check A. That's cooperative multitasking — one chef, many tasks, each yielding control while waiting for I/O.

• Coroutine: a function that can pause (await) and resume later.
• Event loop: the scheduler that decides which coroutine runs next.
• async def: defines an async function.
• await: yields control until the awaited operation completes.

# Sync version: drinks coffee while waiting
import time

def get_cloud_status(name):
    print(f"Querying {name}...")
    time.sleep(2)  # Simulate network call
    return {"name": name, "status": "ok"}

start = time.time()
results = [get_cloud_status(n) for n in ["aws", "azure", "gcp"]]
print(f"Took {time.time() - start:.2f}s")

# Async version: the same, but faster
import asyncio
import time

async def get_cloud_status(name):
    print(f"Querying {name}...")
    await asyncio.sleep(2)  # Simulate network I/O
    return {"name": name, "status": "ok"}

async def main():
    start = time.time()
    results = await asyncio.gather(
        get_cloud_status("aws"),
        get_cloud_status("azure"),
        get_cloud_status("gcp")
    )
    print(f"Took {time.time() - start:.2f}s")

asyncio.run(main())

Notice: the async version completes in ~2 seconds instead of 6. That's the power of overlapping I/O. In real cloud APIs, network latency dominates, so this pattern delivers massive speedups.

How It Works Step by Step

Step 1: Choose Your Async HTTP Client

Most cloud SDKs are synchronous under the hood (e.g., boto3 uses requests). To go async, you have two options:

1. Use an async-native SDK — e.g., aioboto3 (a thin wrapper over boto3), or Azure's azure-mgmt-* with aiohttp-based clients.
2. Call the REST API directly with httpx or aiohttp, handling authentication yourself.

For this lesson, we'll focus on the second approach, because it's universal — you can apply it to any cloud provider that exposes a REST endpoint.

Step 2: Manage Authentication as Async

Authentication often involves tokens (OAuth2 for Azure/GCP, or IAM signing for AWS). You need to fetch and refresh tokens asynchronously. For example, Azure Active Directory tokens can be obtained via httpx in an async function.

Step 3: Build a Reusable Async Client

Create a class that holds the base URL, headers, and session management.

import asyncio
import httpx

class AsyncCloudAPIClient:
    def __init__(self, base_url, token):
        self.base_url = base_url
        self.headers = {"Authorization": f"Bearer {token}"}
        self.client = httpx.AsyncClient(headers=self.headers)

    async def get(self, path):
        resp = await self.client.get(f"{self.base_url}{path}")
        resp.raise_for_status()
        return resp.json()

    async def close(self):
        await self.client.aclose()

Step 4: Define Your Query Functions

Create async functions that use the client to fetch specific data — e.g., list instances, get status, etc.

Step 5: Run Queries Concurrently with asyncio.gather

Use asyncio.gather() to fire off multiple queries at once. Be mindful of rate limits — introduce a semaphore to limit concurrency.

async def main():
    client = AsyncCloudAPIClient("https://api.example.com", token="dummy")
    sem = asyncio.Semaphore(10)  # max 10 concurrent queries

    async def bounded_query(path):
        async with sem:
            return await client.get(path)

    paths = [f"/instances?id={i}" for i in range(50)]
    results = await asyncio.gather(*[bounded_query(p) for p in paths])
    await client.close()
    print("Got", len(results), "responses")

asyncio.run(main())

Hands-On Walkthrough: Async AWS EC2 Inventory

Let's apply this to a real DevOps scenario — fetching EC2 instance status across all regions using async Python and the AWS REST API (or aioboto3).

Install dependencies

pip install httpx aioboto3 boto3

Full example with aioboto3

aioboto3 gives you boto3's API but with async/await support.

import asyncio
import aioboto3
from botocore.config import Config

async def get_ec2_status(session, region):
    async with session.client("ec2", region_name=region) as ec2:
        paginator = ec2.get_paginator("describe_instances")
        instances = []
        async for page in paginator.paginate():
            for reservation in page.get("Reservations", []):
                for inst in reservation.get("Instances", []):
                    instances.append({
                        "id": inst["InstanceId"],
                        "state": inst["State"]["Name"]
                    })
        return region, instances

async def main():
    # Get all regions (cached, so it's fast)
    session = aioboto3.Session()
    async with session.client("ec2", region_name="us-east-1") as ec2:
        regions = [r["RegionName"] for r in (await ec2.describe_regions())["Regions"]]

    tasks = [get_ec2_status(session, r) for r in regions]
    results = await asyncio.gather(*tasks)

    total = 0
    for region, instances in results:
        print(f"{region}: {len(instances)} instances")
        total += len(instances)
    print(f"Total instances: {total}")

asyncio.run(main())

Expected output (abbreviated):

us-east-1: 12 instances
us-west-2: 8 instances
eu-west-1: 5 instances
...
Total instances: 25

Pro tip: Always close the async client to release connections, or use async with context manager.

Compare Options / When to Choose What

When to choose what: - If you're already deep in AWS, use aioboto3 — it's a drop-in replacement. - If you're working with a lesser-known cloud provider, go with httpx REST calls to avoid waiting for SDK support. - For a one-off script, a ThreadPoolExecutor with regular boto3 might be simpler to write and debug.

# ThreadPoolExecutor alternative
from concurrent.futures import ThreadPoolExecutor
import boto3

def get_status(region):
    ec2 = boto3.client("ec2", region_name=region)
    response = ec2.describe_instances()
    count = sum(len(r["Instances"]) for r in response["Reservations"])
    return region, count

with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(get_status, regions))

Troubleshooting & Edge Cases

Problem: "RuntimeError: asyncio.run() cannot be called from a running event loop"

This happens if you're inside a Jupyter notebook or another async context. Solution: use await main() if already in an async environment, or run with nest_asyncio in notebooks.

Problem: Rate limiting (HTTP 429 or 403)

Cloud providers throttle API calls. You'll see errors like TooManyRequestsException (AWS) or 429 (Too Many Requests).

Fix: Add a semaphore to limit concurrency, implement exponential backoff with asyncio.sleep, and consider caching responses.

async def get_with_retry(client, path, max_retries=3):
    for attempt in range(max_retries):
        resp = await client.get(path)
        if resp.status_code == 200:
            return resp.json()
        if resp.status_code == 429:
            await asyncio.sleep(2 ** attempt)
        else:
            resp.raise_for_status()
    raise Exception("Max retries exceeded")

Problem: Authentication token expires mid-run

Tokens are often valid for 30-60 minutes. If you're doing a long batch, refresh tokens asynchronously in the background.

Problem: Unexpected connection errors or timeouts

Always use a timeout on your httpx.AsyncClient to avoid hanging forever. Set timeout=10 or similar.

client = httpx.AsyncClient(timeout=httpx.Timeout(10.0), headers=headers)

What You Learned & What's Next

You now understand how to query cloud APIs with async Python to dramatically speed up your DevOps automation. You learned:

• The difference between synchronous and asynchronous I/O
• How to build an async REST client with httpx
• How to use aioboto3 to query AWS EC2 asynchronously
• How to handle concurrency limits and retries
• When to choose async SDK vs. REST vs. thread pools

These skills apply to any cloud provider and are foundational for building responsive, production-ready automation tools.

Next lesson in this track: we'll build on this foundation to stream and process cloud events in real time using async queues and WebSocket connections. That's where async truly shines.

Take the next step: try modifying the example to query Azure VMs or GCP instances using the same pattern, and time your results.