InventoryAllocation

Scheduling Competitor Rate Shops with Async httpx

Shopping a comp set means firing hundreds of requests at a handful of hosts, and the tension is immediate: too slow and the benchmark is stale before the pricing engine reads it, too fast and the source flags the property’s IP and starts returning 429s or CAPTCHAs. The answer is concurrency that is bounded and paced by construction — many requests in flight overall, but never a detectable burst against any single host. This page builds that scheduler with async httpx, as the collection layer of the competitor rate shopping ingestion workflow. It captures the raw response for every shop so a published benchmark can always be traced back to the exact bytes it came from.

The goal is a collector that saturates your polite budget without exceeding it: a global concurrency ceiling, a separate per-host minimum interval with jitter, retries that respect Retry-After, and an audit capture that is part of the fetch rather than bolted on after.

Prerequisites & environment

The collector sits between the shop matrix and the normalizer. Pin these so pacing behaves identically across workers:

The ShopRequest shape and the benchmark store this feeds are defined on the parent guide; this page owns only the concurrent, polite fetch.

Bounded, per-host-paced concurrent rate-shop scheduling A queue of shop requests on the left feeds a global semaphore that caps total requests in flight at eight. Requests that pass the semaphore then wait at one of three per-host interval gates — one for booking.com, one for expedia, one for a vendor — each enforcing a minimum spacing with jitter so no host sees a burst. Requests that clear a host gate are fetched by a shared async httpx client, wrapped in a tenacity retry that honours Retry-After. Each response's raw bytes are captured to an audit blob before the extracted rate flows on to the normalizer. Concurrent, but polite per host Global semaphore caps total load; per-host interval gates with jitter prevent any single-host burst. Shop queue request matrix Global semaphore ≤ 8 in flight host gate · booking host gate · expedia host gate · vendor httpx client tenacity · Retry-After Audit blob raw bytes captured To normalizer extracted rate
Figure 1: requests clear a global concurrency ceiling, then wait at a per-host interval gate with jitter, then are fetched by a shared httpx client under a tenacity retry. Raw bytes are captured for audit before the extracted rate continues to the normalizer.

Step-by-step implementation

Build the collector in four layers: a per-host interval gate, a tenacity retry policy, a fetch that captures raw bytes, and a bounded scheduler that runs the whole matrix.

Step 1 — Gate each host to a minimum interval with jitter

The global semaphore caps total load, but politeness is per host: two hosts can each take four in-flight requests while neither sees a burst. A per-host gate enforces a minimum spacing between requests to the same host and adds jitter so the arrivals never form a detectable rhythm.

python
import asyncio, random, time
from collections import defaultdict

class HostRateGate:
    """Enforces a minimum interval between requests to the same host, with jitter."""
    def __init__(self, min_interval: float = 1.5, jitter: float = 0.5):
        self._min = min_interval
        self._jitter = jitter
        self._next_allowed: dict[str, float] = defaultdict(float)
        self._locks: dict[str, asyncio.Lock] = defaultdict(asyncio.Lock)

    async def wait(self, host: str) -> None:
        async with self._locks[host]:            # serialize the gate check per host
            now = time.monotonic()
            wait_for = self._next_allowed[host] - now
            if wait_for > 0:
                await asyncio.sleep(wait_for)
            # schedule the next slot: base interval plus a random jitter
            spacing = self._min + random.uniform(0, self._jitter)
            self._next_allowed[host] = time.monotonic() + spacing

Holding a per-host lock while computing and reserving the next slot is what makes the gate correct under concurrency: without it, two coroutines could read the same _next_allowed value and both fire immediately, collapsing the interval and producing exactly the burst the gate exists to prevent.

Step 2 — Define the retry policy once, honouring Retry-After

Wrap the transport call in a tenacity retry so backoff lives in one place. Retry only the transient statuses, cap the attempts, and prefer the source’s own Retry-After when it sends one.

python
import httpx
from tenacity import (retry, stop_after_attempt, wait_exponential_jitter,
                      retry_if_exception_type)

RETRYABLE = {429, 500, 502, 503, 504}

class TransientShopError(Exception):
    """Raised for a retryable HTTP status so tenacity can back off."""

@retry(reraise=True,
       stop=stop_after_attempt(4),
       wait=wait_exponential_jitter(initial=1.0, max=30.0),
       retry=retry_if_exception_type(TransientShopError))
async def _get_with_retry(client: httpx.AsyncClient, url: str) -> httpx.Response:
    resp = await client.get(url, timeout=15.0)
    if resp.status_code in RETRYABLE:
        # a Retry-After header, when present, overrides the backoff floor
        retry_after = resp.headers.get("Retry-After")
        if retry_after and retry_after.isdigit():
            await asyncio.sleep(int(retry_after))
        raise TransientShopError(f"status {resp.status_code}")
    return resp

Raising a custom TransientShopError only for the retryable set — rather than blanket-retrying every non-200 — keeps a terminal 403 or a 404 from being retried, so the client backs off on genuine throttling but fails fast on a block or a dead URL instead of burning the retry budget against a wall.

Step 3 — Fetch one shop and capture the raw response

Each fetch passes the host gate, calls the retrying transport, and persists the raw bytes to the audit sink before returning the extracted fields. Capture is inside the fetch so a response can never reach the normalizer without also reaching the audit trail.

python
import structlog
from urllib.parse import urlparse

log = structlog.get_logger()

async def shop_one(req, client: httpx.AsyncClient, gate: HostRateGate,
                   audit_sink: dict) -> dict | None:
    url = build_source_url(req)             # source adapter maps a ShopRequest to a URL
    host = urlparse(url).netloc
    await gate.wait(host)                    # politeness: block until this host's slot opens
    try:
        resp = await _get_with_retry(client, url)
    except (TransientShopError, httpx.HTTPError) as exc:
        log.warning("shop_failed", competitor=req.competitor_property_id, error=str(exc))
        return None
    if resp.status_code == 403:              # flagged: terminal for this host, do not retry
        log.error("shop_forbidden", competitor=req.competitor_property_id, host=host)
        return None
    audit_sink[f"{host}:{req.competitor_property_id}:{req.check_in}"] = resp.text
    log.info("shop_ok", competitor=req.competitor_property_id, host=host,
             status=resp.status_code, bytes=len(resp.content))
    return extract_rate_fields(resp)         # adapter pulls amount/currency/board out

Writing resp.text to the audit sink before extraction — not after — guarantees that even if extract_rate_fields later raises on a changed page layout, the raw bytes are already captured, so the failing response is available to debug the adapter rather than lost.

Step 4 — Run the matrix under a bounded scheduler

Drive the whole shop matrix through a global semaphore so total concurrency is capped regardless of how many hosts are in play, using a TaskGroup so a crash in one shop does not silently swallow the rest.

python
async def run_shops(matrix: list, client: httpx.AsyncClient,
                    gate: HostRateGate, audit_sink: dict,
                    max_in_flight: int = 8) -> list[dict]:
    sem = asyncio.Semaphore(max_in_flight)
    results: list[dict] = []

    async def _bounded(req):
        async with sem:                      # global ceiling across all hosts
            out = await shop_one(req, client, gate, audit_sink)
            if out is not None:
                results.append(out)

    async with asyncio.TaskGroup() as tg:    # structured concurrency: all-or-surface
        for req in matrix:
            tg.create_task(_bounded(req))
    log.info("shop_run_complete", planned=len(matrix), collected=len(results))
    return results

Separating the global semaphore from the per-host gate is the key design choice: the semaphore protects your worker and the shared rate-limit budget from too many open sockets, while the gate protects each source from a burst — one ceiling cannot serve both jobs, because eight concurrent requests spread across four hosts is polite while eight against one host is an attack.

Gotchas & production notes

Verification snippet

Confirm the two politeness guarantees — that per-host spacing actually holds and that the global ceiling is never exceeded — with a fake clock and counter before pointing the collector at live sources.

python
import asyncio, time

async def test_host_gate_enforces_spacing():
    gate = HostRateGate(min_interval=0.2, jitter=0.0)
    start = time.monotonic()
    # three sequential waits on one host must span at least 2 * min_interval
    for _ in range(3):
        await gate.wait("booking.com")
    elapsed = time.monotonic() - start
    assert elapsed >= 0.4                     # first is free, next two spaced 0.2s each

async def test_global_ceiling_never_exceeded():
    peak = 0
    live = 0
    sem = asyncio.Semaphore(3)

    async def worker():
        nonlocal peak, live
        async with sem:
            live += 1
            peak = max(peak, live)
            await asyncio.sleep(0.01)
            live -= 1

    async with asyncio.TaskGroup() as tg:
        for _ in range(20):
            tg.create_task(worker())
    assert peak <= 3                          # semaphore held the ceiling under 20 tasks

asyncio.run(test_host_gate_enforces_spacing())
asyncio.run(test_global_ceiling_never_exceeded())

The first test proves the per-host gate actually spaces requests rather than merely intending to; the second proves the semaphore holds the global ceiling even when far more tasks are queued than slots exist. Together they are the two properties that keep a comp-set shop polite. In production, also assert that the audit sink holds one raw entry per successful fetch, so a gap between planned shops and captured responses surfaces a silent collection failure.

← Back to Competitor Rate Shopping Ingestion