Async Polling for Inventory Updates: Deterministic Rate Parity in Hotel Distribution

Asynchronous polling is the synchronization mechanism of last resort — and, in practice, of daily reality — for hotel property management systems (PMS) and channel managers whenever push delivery is unreliable, silently dropped, or simply unsupported by a legacy OTA endpoint. When channel manager webhook delivery misses an availability event, the failure is invisible: no error is raised, the room quietly stays sellable, and the first signal a revenue manager gets is an overbooking or a parity violation flagged by Booking.com. A polling workflow removes that blind spot by continuously pulling the current inventory state, comparing it against a local baseline, and pushing corrective deltas into the pricing engine on a predictable cadence. This page sits inside the broader API Sync & Data Ingestion Workflows domain and specifies the exact architecture, code, data contracts, and failure handling a Python automation engineer needs to run polling in production for revenue and operations teams.

Unlike event-driven designs that assume perfect message delivery, polling enforces strict idempotency, explicit state tracking, and bounded latency. Revenue managers depend on it to hold rate parity across fragmented distribution channels; operations teams depend on it to prevent phantom availability; engineers depend on it because it degrades gracefully when an upstream system misbehaves.

Architecture & Prerequisites

Polling for inventory updates functions as a scheduled ingestion pipeline that bridges real-time pricing decisions and asynchronous channel responses. A centralized scheduler triggers concurrent polling jobs scoped per property_id, per channel, and per room_type_code. Each job holds a local state cache containing the last known availability, rate_plan_code, and restriction matrix. On each tick, the job issues an authenticated GET to the channel manager or OTA inventory endpoint, retrieves the current payload, validates it against the canonical schema, and runs a deterministic diff against the cached baseline. Only validated deltas proceed to the rate parity engine, which pushes corrective updates downstream. This isolates transient network failures from core pricing logic so revenue operations never act on stale or partially synced inventory.

Environment assumptions and dependency versions:

• Python 3.11+ (for asyncio.TaskGroup and improved exception groups)
• httpx 0.27+ for non-blocking HTTP/2-capable I/O
• tenacity 8.x for declarative retry policies
• pydantic 2.6+ (v2 API — field_validator, model_dump) for payload contracts
• structlog 24.x for JSON-structured telemetry
• redis 5.x (optional) for a shared, restart-durable state cache
• Read access to a channel manager or OTA inventory endpoint, with credentials managed through OAuth2 token refresh so a mid-poll 401 never strands a cycle

The polling engine consumes the same canonical inventory shape produced by standardized JSON payloads, and the rate_plan_code values it compares must already conform to your rate plan taxonomy. Polling upstream of those two contracts is possible but fragile: unnormalized rate plans produce false deltas on every cycle.

Implementation

The build proceeds in four numbered steps. Each step is anchored to a self-contained, runnable block; wire them together in the order shown.

Step 1 — Define the polling engine and resilient fetch

The fetch method is the only part of the loop that touches the network, so it is the only part wrapped in a retry policy. Keeping retries at this granularity means a transient failure re-issues one request rather than replaying the whole cycle and re-emitting deltas.

import asyncio
from typing import Any, Optional

import httpx
import structlog
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
)

structlog.configure(processors=[structlog.processors.JSONRenderer()])
log = structlog.get_logger()


class PollingEngine:
    def __init__(
        self,
        base_url: str,
        auth_headers: dict[str, str],
        state_cache: dict[str, "InventorySnapshot"],
    ) -> None:
        self.base_url = base_url
        self.auth_headers = auth_headers
        self.state_cache = state_cache
        self.client = httpx.AsyncClient(timeout=15.0, http2=True)

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type((httpx.RequestError, httpx.HTTPStatusError)),
        reraise=True,
    )
    async def fetch_inventory(self, property_id: str, room_type_code: str) -> dict[str, Any]:
        endpoint = f"{self.base_url}/inventory/{property_id}/rooms/{room_type_code}"
        response = await self.client.get(endpoint, headers=self.auth_headers)
        response.raise_for_status()
        return response.json()

The retry_if_exception_type guard deliberately retries only network and HTTP status errors, so a schema or logic bug raises immediately instead of being silently masked by three attempts.

Step 2 — Compute a deterministic delta

The diff keys off a monotonic version_token rather than a naive field comparison. If the token is unchanged the payload is equivalent to the baseline and the cycle exits early, which is what keeps a stable channel from generating pointless parity pushes.

    def compute_delta(
        self, cached: "InventorySnapshot", live: dict[str, Any]
    ) -> Optional[dict[str, Any]]:
        try:
            current = InventorySnapshot(**live)
        except ValidationError as exc:
            log.error("schema_validation_failed", error=str(exc), payload=live)
            return None

        if current.version_token == cached.version_token:
            return None
        if current.sequence_no <= cached.sequence_no:
            log.warning(
                "stale_payload_discarded",
                property_id=current.property_id,
                room_type_code=current.room_type_code,
                cached_seq=cached.sequence_no,
                live_seq=current.sequence_no,
            )
            return None

        return {
            "property_id": current.property_id,
            "room_type_code": current.room_type_code,
            "rate_plan_code": current.rate_plan_code,
            "availability_change": current.available_units - cached.available_units,
            "rate_plan_changed": current.rate_plan_code != cached.rate_plan_code,
            "new_version_token": current.version_token,
        }

The sequence_no guard rejects out-of-order responses — a real hazard when concurrent OTA replicas answer the same poll window — so the cache only ever moves forward in time.

Step 3 — Apply the correction idempotently

The downstream push carries an idempotency key derived from the property, room, and version token. Replaying the same delta (for example after a scheduler restart) is therefore a no-op at the parity engine instead of a duplicate rate write.

    async def apply_parity_correction(self, delta: dict[str, Any]) -> None:
        idempotency_key = (
            f"{delta['property_id']}:{delta['room_type_code']}:{delta['new_version_token']}"
        )
        await self.client.post(
            f"{self.base_url}/parity/corrections",
            headers={**self.auth_headers, "Idempotency-Key": idempotency_key},
            json=delta,
        )
        log.info("parity_delta_applied", idempotency_key=idempotency_key, **delta)

Deriving the key from version_token rather than a timestamp guarantees that two workers processing the same upstream state produce the identical key and collapse into a single correction.

Step 4 — Run the concurrent scheduler

asyncio.gather with return_exceptions=True lets one property’s failure fall through to structured logging without aborting the other jobs in the same tick. The engine polls hundreds of property/room pairs concurrently on a single event loop, with no thread overhead.

    async def poll_cycle(self, property_id: str, room_type_code: str) -> None:
        cache_key = f"{property_id}:{room_type_code}"
        cached = self.state_cache.get(cache_key)
        if cached is None:
            log.warning("cache_miss_initializing", property_id=property_id, room_type_code=room_type_code)
            return
        try:
            payload = await self.fetch_inventory(property_id, room_type_code)
            delta = self.compute_delta(cached, payload)
            if delta:
                await self.apply_parity_correction(delta)
                self.state_cache[cache_key] = InventorySnapshot(**payload)
                log.info("inventory_synced", property_id=property_id,
                         room_type_code=room_type_code, version=payload["version_token"])
        except httpx.HTTPStatusError as exc:
            log.error("ota_http_error", status=exc.response.status_code, url=str(exc.request.url))
        except Exception:
            log.exception("polling_unexpected_failure", property_id=property_id, room_type_code=room_type_code)

    async def run_scheduler(self, job_configs: list[dict], interval: int = 60) -> None:
        while True:
            tasks = [self.poll_cycle(c["property_id"], c["room_type_code"]) for c in job_configs]
            await asyncio.gather(*tasks, return_exceptions=True)
            await asyncio.sleep(interval)

Because the whole cycle is bounded by the 15-second client timeout and a three-attempt retry budget, the worst-case tick duration is predictable — a prerequisite for choosing a safe interval.

Adaptive cadence. A fixed interval wastes quota during quiet periods and starves freshness during peaks. Production engines drive the interval from booking velocity and seasonality flags: contract to 60–90 seconds through high-demand windows to catch rapid depletion, and expand to 5–10 minutes on shoulder dates to conserve compute and OTA quota. The ceiling on aggressiveness is always the channel’s published request budget, covered under OTA API rate limits.

Schema & Data Contracts

Every payload is validated before it can mutate the cache. The canonical snapshot is a Pydantic v2 model; field_validator enforces the invariants that a raw OTA response routinely violates (negative availability, unmapped rate plans, lower-cased channel slugs).

from datetime import datetime
from pydantic import BaseModel, Field, ValidationError, field_validator

ALLOWED_CHANNELS = {"booking_com", "expedia", "agoda", "direct"}


class InventorySnapshot(BaseModel):
    property_id: str = Field(pattern=r"^prop_[0-9a-f]{8}$")
    room_type_code: str
    rate_plan_code: str
    channel: str
    available_units: int = Field(ge=0)
    closed_to_arrival: bool = False
    min_length_of_stay: int = Field(ge=1, default=1)
    last_updated: datetime
    sequence_no: int = Field(ge=0)
    version_token: str

    @field_validator("channel")
    @classmethod
    def channel_must_be_known(cls, value: str) -> str:
        slug = value.strip().lower()
        if slug not in ALLOWED_CHANNELS:
            raise ValueError(f"unknown channel slug: {value!r}")
        return slug

    @field_validator("rate_plan_code")
    @classmethod
    def rate_plan_uppercase(cls, value: str) -> str:
        return value.strip().upper()

Normalizing channel and rate_plan_code inside the model — rather than at the diff site — means compute_delta compares already-canonical values and never mis-fires a delta on a formatting difference alone. Serialize with snapshot.model_dump(mode="json") when persisting to Redis so datetime fields round-trip as ISO-8601 strings.

Error Handling & Retry Strategy

Polling faces a small, well-defined set of upstream failure classes, and each maps to a distinct action. Reuse the shared taxonomy in categorizing 4xx vs 5xx sync errors rather than inventing per-endpoint logic.

The retry ceiling for 429 and 5xx uses the same exponential backoff profile as the rest of the pipeline: wait_exponential(multiplier=1, min=2, max=10) capped at three attempts, with full jitter to avoid a synchronized retry stampede across properties. The idempotency key design in Step 3 is what makes retries safe: because the key is deterministic in version_token, a 429-then-retry that actually succeeded on the first (throttled) attempt collapses to a 409 on replay instead of writing the rate twice.

When drift exceeds the configured tolerance — for example, more than a handful of unit changes accumulate while the endpoint was throttled — the engine should stop issuing piecemeal corrections and escalate the whole property to batch reconciliation. Piecemeal catch-up under load is exactly what trips OTA parity-compliance flags.

Verification & Testing

Confirming a polling deployment is behaving means asserting on both the log stream and the cache state, not just a green process.

import asyncio

async def test_delta_is_idempotent() -> None:
    baseline = InventorySnapshot(
        property_id="prop_0a1b2c3d", room_type_code="DLX_KING",
        rate_plan_code="BAR", channel="booking_com", available_units=5,
        last_updated="2026-07-02T09:00:00Z", sequence_no=41, version_token="v41",
    )
    cache = {"prop_0a1b2c3d:DLX_KING": baseline}
    engine = PollingEngine("https://cm.example", {"Authorization": "Bearer t"}, cache)

    live = baseline.model_dump(mode="json") | {
        "available_units": 3, "sequence_no": 42, "version_token": "v42",
    }
    delta = engine.compute_delta(baseline, live)
    assert delta["availability_change"] == -2
    # Replaying the identical payload after cache update yields no delta.
    engine.state_cache["prop_0a1b2c3d:DLX_KING"] = InventorySnapshot(**live)
    assert engine.compute_delta(InventorySnapshot(**live), live) is None

asyncio.run(test_delta_is_idempotent())

This smoke test proves the two properties that matter most: a real change produces the correct signed availability_change, and re-observing the same state produces nothing. In production, additionally assert that the count of parity_delta_applied log events equals the count of distinct version_token values seen per property per hour — any excess means the diff is mis-firing. Track poll_duration_ms, delta_magnitude, retry_count, and cache_hit_ratio as structured fields to feed observability dashboards and alert before a parity breach reaches booking conversion.

Troubleshooting

Duplicate parity corrections on every cycle : Root cause: rate_plan_code or channel arrives unnormalized, so compute_delta sees a “change” that is only a formatting difference. Fix: ensure the field_validator normalization runs (see the schema section) and that upstream payloads already match your rate plan taxonomy.

Availability lags real bookings during peak windows : Root cause: fixed interval too coarse for booking velocity. Fix: switch to adaptive cadence keyed on booking velocity and seasonality; contract to 60–90 s in peak windows.

Cycles silently stop after several hours : Root cause: the access token expired and the 401 path was never wired, so fetch_inventory exhausted retries and the job died quietly. Fix: intercept 401, invoke OAuth2 token refresh, and resume without dropping the tick.

429 storms across many properties at once : Root cause: all jobs share one API key and retry in lockstep. Fix: add full jitter to backoff, shed concurrency on 429, and track a sliding-window request budget per key.

Only part of a large property’s inventory updates : Root cause: the endpoint paginates across date ranges or room categories and the loop reads only the first page. Fix: apply cursor-based traversal per parsing paginated OTA responses.

FAQ

Related

• Parsing Paginated OTA Responses with Requests — cursor traversal so a poll never reads a partial inventory set
• Handling OTA API Rate Limits — the request budget that caps how aggressively you can poll
• OAuth2 Token Refresh Strategies — keeping credentials valid so a cycle never dies on a 401
• Batch Reconciliation Workflows — the escalation path when drift exceeds tolerance
• Categorizing 4xx vs 5xx Sync Errors — the shared taxonomy behind the retry table above

← Back to API Sync & Data Ingestion Workflows