Async web requests

Theo's Monday quotes are the real test of everything in the last two lessons. The shipping company offers a web API: send an HTTP request with the parcel's weight and destination, receive a JSON reply with prices. The company's documentation sets two rules: no more than ten requests in flight at once per account, and clients must back off and retry when the server replies 429 Too Many Requests or fails temporarily. A good async quote client must therefore run requests concurrently, respect the limit, retry sensibly, survive individual failures, and still finish 200 quotes in under 15 seconds.

This lesson builds that client step by step. It introduces async HTTP libraries such as httpx and aiohttp and why ordinary requests cannot be used inside a coroutine, sharing one client session for many requests, limiting concurrency with asyncio.Semaphore, collecting results and errors with gather and return_exceptions, retrying with exponential backoff and jitter, and processing results as they arrive. The browser cannot make real network requests from these exercises, so the network is replaced by small fake coroutines, and the code you write is the same logic a real client needs.

Async HTTP clients

The popular requests library is blocking: requests.get() stops the thread until the reply arrives. Inside a coroutine that would freeze the event loop, so async programs use HTTP libraries built for asyncio. httpx offers an AsyncClient whose methods are coroutines, with an interface close to requests. aiohttp is an older, widely used async client and server library. Both are third-party packages installed with pip. Both encourage you to create one client or session and reuse it for every request, because the client keeps connections to the server open between requests, saving the cost of opening a new connection and negotiating encryption each time.

import asyncio
import httpx

API = "https://api.example-shipping.test/v2/quotes"

async def get_quote(client, order):
    reply = await client.post(API, json={"weight_kg": order["kg"], "postal": order["postal"]})
    reply.raise_for_status()                    # turns 4xx and 5xx replies into exceptions
    return order["id"], reply.json()["price"]

async def main(orders):
    async with httpx.AsyncClient(timeout=10.0) as client:     # one client, reused
        results = await asyncio.gather(*(get_quote(client, o) for o in orders))
    return dict(results)

orders = [{"id": "A-1001", "kg": 0.68, "postal": "V1Y 2B3"}, {"id": "A-1002", "kg": 1.36, "postal": "T5J 0N3"}]
print(asyncio.run(main(orders)))

This version has the right shape but follows none of the shipping company's rules. With 200 orders, gather starts all 200 requests at once, the server answers most of them with 429 Too Many Requests, and raise_for_status turns those into exceptions. Worse, gather raises the first exception it sees and the results of the successful requests are lost. Each of these problems has a standard fix.

Limiting concurrency with a semaphore

A semaphore is a counter that controls how many tasks may enter a section of code at the same time. asyncio.Semaphore(10) starts at ten. Each task that enters with async with semaphore: decreases the counter, and if the counter is zero the task waits until another task leaves and increases it again. Wrapping the request in a semaphore caps the number in flight, while gather still schedules every coroutine at once. The tasks beyond the limit simply wait their turn, costing almost nothing while they wait.

import asyncio, time

in_flight = 0
peak = 0

async def get_quote(order_id, limit):
    global in_flight, peak
    async with limit:                         # at most N inside at once
        in_flight += 1
        peak = max(peak, in_flight)
        await asyncio.sleep(0.5)              # the request
        in_flight -= 1
        return order_id, 12.40

async def main():
    limit = asyncio.Semaphore(10)
    orders = [f"A-{1000 + n}" for n in range(200)]
    start = time.perf_counter()
    quotes = await asyncio.gather(*(get_quote(o, limit) for o in orders))
    print(f"{len(quotes)} quotes in {time.perf_counter() - start:.1f} s; peak in flight: {peak}")

asyncio.run(main())

200 quotes in 10.0 s; peak in flight: 10

The arithmetic is the same as for the thread pool in lesson 6: 200 requests, ten at a time, half a second each, gives twenty rounds and about ten seconds, inside the goal. The semaphore belongs to the event loop, so it should be created inside a coroutine, as main does here, and shared by passing it to the tasks.

Collecting results and errors

By default, asyncio.gather raises the first exception from any of its coroutines, and you lose access to the other results. Passing return_exceptions=True changes that: gather waits for everything and returns a list in which each entry is either a result or the exception that coroutine raised. The program can then separate successes from failures, report the failures and retry them later. For long jobs, processing results as they arrive with asyncio.as_completed gives early feedback, much like the futures version in lesson 6.

def drive(coro):
    """Run a coroutine to completion by stepping it (enough for await asyncio.sleep(0))."""
    try:
        while True:
            coro.send(None)
    except StopIteration as done:
        return done.value

import asyncio

async def get_quote(order_id):
    await asyncio.sleep(0)
    if order_id.endswith("3"):
        raise ConnectionError(f"timeout for {order_id}")
    return 12.40

async def main(orders):
    results = await asyncio.gather(*(get_quote(o) for o in orders), return_exceptions=True)
    ok = {o: r for o, r in zip(orders, results) if not isinstance(r, Exception)}
    bad = {o: str(r) for o, r in zip(orders, results) if isinstance(r, Exception)}
    return ok, bad

orders = ["A-1001", "A-1002", "A-1003", "A-1004"]
coros = [get_quote(o) for o in orders]
ok, bad = {}, {}
for o, c in zip(orders, coros):                  # stepped one by one here, for the browser
    try:
        ok[o] = drive(c)
    except ConnectionError as e:
        bad[o] = str(e)
print("ok:", ok)
print("failed:", bad)

ok: {'A-1001': 12.4, 'A-1002': 12.4, 'A-1004': 12.4}
failed: {'A-1003': 'timeout for A-1003'}

def split_results(order_ids, results):
    return dict(zip(order_ids, results)), {}

print(split_results(['A-1', 'A-2'], [12.4, ConnectionError('timeout')]))

Retrying with exponential backoff

Some failures are permanent, such as a postal code that does not exist, which returns 400 Bad Request; retrying will never help. Others are temporary: a 429 when the server is busy, a 503 Service Unavailable, a dropped connection. For temporary failures the right response is to wait and try again, but not immediately and not all at the same moment, or every client hammers the struggling server in lockstep. The standard approach is exponential backoff: wait a base delay after the first failure, then double the wait after each further failure, up to a cap. Adding jitter, a small random variation in each delay, stops many clients from retrying in sync.

import random

def backoff_delays(attempts, base=0.5, cap=8.0):
    """Delays before each retry: base, 2*base, 4*base, ... capped."""
    return [min(cap, base * 2 ** i) for i in range(attempts)]

def with_jitter(delays, seed=1):
    rnd = random.Random(seed)
    return [round(d * rnd.uniform(0.5, 1.0), 2) for d in delays]   # "equal jitter" style

print("backoff:    ", backoff_delays(6))
print("with jitter:", with_jitter(backoff_delays(6)))
print("total worst-case wait:", sum(backoff_delays(6)), "s")

backoff:     [0.5, 1.0, 2.0, 4.0, 8.0, 8.0]
with jitter: [0.28, 0.92, 1.76, 2.51, 5.98, 5.8]
total worst-case wait: 23.5 s

When a server sends a Retry-After header with a 429 or 503 reply, a polite client waits at least that long instead of its own calculated delay. A client should also give up after a fixed number of attempts and report the failure, rather than retrying forever. In async code the retry loop wraps the awaited request, and await asyncio.sleep(delay) waits without blocking other tasks.

import asyncio

async def retry(make_request, attempts):
    return await make_request()

print(drive(retry(flaky(0), 3)))

Putting the client together

The finished client combines the pieces: one shared AsyncClient, a semaphore of ten, a retry loop with exponential backoff and jitter for temporary errors, and gather with return_exceptions so a few failures do not lose the rest. Each piece is small, and together they make a client that is fast, polite to the server and robust. Theo's Monday run now finishes in about eleven seconds, including a handful of retries, and prints a short list of orders that need attention.

import asyncio, random
import httpx

API = "https://api.example-shipping.test/v2/quotes"
TEMPORARY = {429, 500, 502, 503, 504}

async def get_quote(client, limit, order, attempts=5):
    for attempt in range(attempts):
        async with limit:
            reply = await client.post(API, json={"weight_kg": order["kg"], "postal": order["postal"]})
        if reply.status_code not in TEMPORARY:
            reply.raise_for_status()                      # permanent errors are raised at once
            return reply.json()["price"]
        wait = float(reply.headers.get("Retry-After", min(8.0, 0.5 * 2 ** attempt)))
        await asyncio.sleep(wait * random.uniform(0.5, 1.0))
    raise ConnectionError(f"{order['id']}: gave up after {attempts} tries")

async def quote_all(orders):
    limit = asyncio.Semaphore(10)
    async with httpx.AsyncClient(timeout=10.0) as client:
        results = await asyncio.gather(*(get_quote(client, limit, o) for o in orders),
                                       return_exceptions=True)
    quotes = {o["id"]: r for o, r in zip(orders, results) if not isinstance(r, Exception)}
    failed = {o["id"]: repr(r) for o, r in zip(orders, results) if isinstance(r, Exception)}
    return quotes, failed

The semaphore is released while the client sleeps between retries, because the sleep sits outside the async with block. A retrying request should not hold one of the ten slots while it waits, or a burst of failures could stall the whole job.

Rate limits over time

A concurrency limit caps how many requests are in flight. Many APIs instead limit how many requests may start per second or per minute, a rate limit. The two are different: ten fast requests in flight could still exceed a limit of five requests per second. A simple way to respect a rate limit is to space out request starts so that at most n begin in any one-second window; libraries such as aiolimiter implement the token bucket algorithm for this. The planning arithmetic is straightforward, and worth doing before the job runs.

import math

def job_time(n_requests, per_second=None, concurrent=None, request_s=0.5):
    """Lower bound on time from a rate limit and a concurrency limit."""
    by_rate = n_requests / per_second if per_second else 0
    by_concurrency = math.ceil(n_requests / concurrent) * request_s if concurrent else n_requests * request_s
    return max(by_rate, by_concurrency)

print("200 quotes, 10 in flight:", job_time(200, concurrent=10), "s")
print("200 quotes, 10 in flight, 15 per second:", round(job_time(200, per_second=15, concurrent=10), 1), "s")
print("200 quotes, 50 in flight, 15 per second:", round(job_time(200, per_second=15, concurrent=50), 1), "s")

200 quotes, 10 in flight: 10.0 s
200 quotes, 10 in flight, 15 per second: 13.3 s
200 quotes, 50 in flight, 15 per second: 13.3 s

def backoff(attempts, base, cap):
    return [base] * attempts

def total_wait(attempts, base, cap):
    return 0

print(backoff(5, 0.5, 4))