Data Ingestion Recipe Parsing Workflows

Rate Limiting Strategies for POS APIs

This page shows a Python automation engineer how to build a header-aware rate limiter that keeps a multi-unit POS polling loop under the vendor’s throughput ceiling — the specific task of never triggering an HTTP 429 while still draining transaction deltas fast enough to feed the food-cost engine. It sits under POS API Polling Strategies, narrowing that module’s general delta-sync loop to the throttling, backoff, and quota-accounting concerns that decide whether ingestion stays deterministic at fleet scale.

POS vendors cap throughput in requests per minute (RPM) or concurrent connections to protect their transactional databases. Breach the ceiling and you get 429 Too Many Requests, temporary IP blocks, or — worst of all — silently truncated payloads that corrupt cost-per-portion math downstream. The limiter below enforces the ceiling client-side, honours the vendor’s own Retry-After signal, and validates that every batch arrived complete before it reaches the theoretical food cost calculation from BOMs.

Request lifecycle of a header-aware POS rate limiter A poll call is gated by a sliding-window deque (60-second window) and a concurrency semaphore, dispatched via aiohttp GET, then classified by HTTP status. On 200 the batch is checked for per-store sequence gaps and handed to the cost engine. On 429 or 5xx the call parses Retry-After (delta-seconds or HTTP-date) or falls back to full-jitter exponential backoff and loops back to the gate. After N consecutive failures the circuit breaker opens for a cooldown, and ingestion falls back to CSV reload. 1 · ENQUEUE 2 · WINDOW GATE 3 · DISPATCH 4 · CLASSIFY 200 OK 429 / 5xx retry gate ≥ N fails complete Poll call delta cursor Sliding-window gate deque · 60 s window + concurrency semaphore aiohttp GET session.get() HTTP status 200 · 429 · 5xx Validate completeness per-store seq diff Backoff & retry wait = Retry-After (Δs · date) else full-jitter 2ⁿ backoff Circuit open · cooldown pause endpoint → CSV reload fallback Cost engine handoff clean transaction deltas only

Prerequisites and Data Contract

Pin these versions and provision the quota config before the steps apply. The limiter is only deterministic if the ceiling it enforces matches the ceiling the vendor actually enforces.

  • Runtime: Python 3.11+, aiohttp==3.9.*, pydantic==2.7.*, pandas==2.2.*.
  • Concurrency model: a single asyncio event loop per worker process. The limiter is shared by all coroutines in that loop; do not share one instance across OS processes — give each process its own budget slice.
  • Environment: POS_API_KEY, POS_BASE_URL, and a quota profile per vendor/tier. Store the profile in config, never hardcode it inline, so a subscription upgrade is a data change rather than a deploy.

The quota profile is a fixed contract. Each POS vendor tier maps to one row:

Field Type Meaning
vendor text POS vendor key (e.g. toast, square)
tier text Subscription tier the RPM applies to
max_requests int Requests allowed per window
window_seconds int Rolling window length (usually 60)
max_concurrency int Simultaneous in-flight requests allowed
safety_margin float Fraction of the ceiling to actually use (0.85)

The safety_margin matters: publish 85–90% of the documented ceiling so clock skew and co-tenant traffic from other pollers on the same account do not push you over. Vendor SKU labels arriving in each payload are assumed already reconciled against your POS taxonomy mapping; the limiter only governs transport, not semantics.

Step-by-Step Implementation

Each step is a self-contained block. Compose them in order inside one polling worker.

Step 1 — Model the quota as a typed config

Load the profile into a frozen Pydantic model so an invalid ceiling fails at startup, not at 2 a.m. during Friday dinner service. The effective_rpm property applies the safety margin once, centrally.

from pydantic import BaseModel, Field, PositiveInt


class QuotaProfile(BaseModel, frozen=True):
    vendor: str
    tier: str
    max_requests: PositiveInt
    window_seconds: PositiveInt = 60
    max_concurrency: PositiveInt = 4
    safety_margin: float = Field(default=0.85, gt=0.0, le=1.0)

    @property
    def effective_rpm(self) -> int:
        # Never round up past the documented ceiling.
        return max(1, int(self.max_requests * self.safety_margin))

Step 2 — Build the sliding-window gate

The gate keeps a deque of monotonic send timestamps. Before each dispatch it prunes expired entries, and if the window is full it sleeps exactly until the oldest entry ages out. A semaphore caps simultaneous in-flight calls so a slow vendor cannot let concurrency balloon past the tier limit.

import asyncio
import time
from collections import deque


class SlidingWindowLimiter:
    def __init__(self, quota: QuotaProfile) -> None:
        self._quota = quota
        self._sent: deque[float] = deque()
        self._lock = asyncio.Lock()
        self._slots = asyncio.Semaphore(quota.max_concurrency)

    async def acquire(self) -> None:
        await self._slots.acquire()
        async with self._lock:
            window = self._quota.window_seconds
            while True:
                now = time.monotonic()
                while self._sent and self._sent[0] <= now - window:
                    self._sent.popleft()
                if len(self._sent) < self._quota.effective_rpm:
                    self._sent.append(now)
                    return
                sleep_for = self._sent[0] + window - now
                await asyncio.sleep(max(sleep_for, 0.0))

    def release(self) -> None:
        self._slots.release()

Step 3 — Parse Retry-After and compute jittered backoff

Per RFC 9110 §10.2.3, Retry-After is either delta-seconds or an HTTP-date — a bare float() breaks on the date form. Parse both, then fall back to exponential backoff with full jitter to stop a fleet of workers retrying in lockstep.

import random
from email.utils import parsedate_to_datetime
from datetime import datetime, timezone


def parse_retry_after(header: str | None) -> float | None:
    if not header:
        return None
    header = header.strip()
    if header.isdigit():
        return float(header)
    try:
        when = parsedate_to_datetime(header)
    except (TypeError, ValueError):
        return None
    if when.tzinfo is None:
        when = when.replace(tzinfo=timezone.utc)
    return max(0.0, (when - datetime.now(timezone.utc)).total_seconds())


def backoff_seconds(attempt: int, retry_after: float | None, cap: float = 30.0) -> float:
    if retry_after is not None:
        return min(retry_after, cap)
    # Full jitter over an exponentially growing window.
    ceiling = min(cap, 0.5 * 2 ** attempt)
    return random.uniform(0.0, ceiling)

Step 4 — Wrap the request with 429 handling

Gate every call through the limiter, then retry on 429 and transient 5xx. The vendor’s Retry-After always wins over the local backoff curve; only when it is absent does the jittered fallback apply.

import aiohttp


async def fetch_deltas(
    session: aiohttp.ClientSession,
    url: str,
    limiter: SlidingWindowLimiter,
    max_retries: int = 5,
) -> dict:
    for attempt in range(max_retries + 1):
        await limiter.acquire()
        try:
            async with session.get(url) as resp:
                if resp.status == 200:
                    return await resp.json()
                if resp.status == 429 or resp.status >= 500:
                    wait = backoff_seconds(attempt, parse_retry_after(resp.headers.get("Retry-After")))
                    await asyncio.sleep(wait)
                    continue
                resp.raise_for_status()
        finally:
            limiter.release()
    raise RuntimeError(f"exhausted {max_retries} retries for {url}")

Step 5 — Trip a circuit breaker on sustained failure

When a vendor endpoint degrades, retrying only deepens the outage and burns quota. A breaker counts consecutive failures, opens after a threshold, and refuses calls for a cooldown so the worker can fall back to a batch reload path instead of hammering a dead API.

from dataclasses import dataclass, field


@dataclass
class CircuitBreaker:
    threshold: int = 5
    cooldown_seconds: float = 120.0
    _failures: int = 0
    _opened_at: float | None = field(default=None)

    def allow(self) -> bool:
        if self._opened_at is None:
            return True
        if time.monotonic() - self._opened_at >= self.cooldown_seconds:
            self._opened_at = None
            self._failures = 0
            return True
        return False

    def record(self, *, ok: bool) -> None:
        if ok:
            self._failures = 0
            self._opened_at = None
        else:
            self._failures += 1
            if self._failures >= self.threshold:
                self._opened_at = time.monotonic()

Step 6 — Validate batch completeness before handoff

Rate limiting keeps the transport clean, but silent truncation still corrupts analytics. Each vendor row carries a per-store monotonic seq; a vectorized groupby(...).diff() flags any gap without row-by-row iteration. Only complete batches move on to the cost engine.

import pandas as pd


def find_sequence_gaps(raw: list[dict], expected_stores: set[str]) -> pd.DataFrame:
    df = pd.DataFrame(raw)
    required = {"store_id", "transaction_id", "timestamp", "seq"}
    missing = required - set(df.columns)
    if missing:
        raise ValueError(f"schema violation, missing columns: {missing}")

    df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True)
    df = df[df["store_id"].isin(expected_stores)].sort_values(["store_id", "seq"])

    # A clean stream increments seq by exactly 1 within each store.
    df["gap"] = df.groupby("store_id")["seq"].diff()
    return df[df["gap"] > 1]

For high-volume backfills or fan-out across hundreds of locations, hand the validated frame to the async batch processing workflow rather than draining every store inline on one loop.

Verification and Validation

Confirm the limiter behaves before you trust a night of unattended polling.

  • Ceiling never exceeded. Assert the effective RPM is honoured under load by counting sends within any window:

    assert len(limiter._sent) <= quota.effective_rpm, "window overflow"

  • Zero 429s in steady state. After a full polling shift, the count of 429 responses should be zero when the endpoint is healthy. Emit a structured log line per cycle and check it:

    logging.info("cycle done status=%s sent=%s waited=%.2fs", "ok", len(items), waited)

  • Gap detection fires. Feed a synthetic batch with a deliberate hole (seq jumping 4 → 6) and confirm find_sequence_gaps returns exactly that store, so truncation cannot slip through silently.

A healthy shift ends with zero 429s, an empty gap frame, and the circuit breaker never opening.

Gotchas and Edge Cases

  • Retry-After as an HTTP-date. Some vendors return Wed, 01 Jul 2026 04:15:00 GMT, not 120. A bare float() raises and the retry path dies mid-loop; parse_retry_after handles both forms.
  • Synchronized retry storms. Fixed backoff makes every worker retry at the same instant, re-triggering the ceiling. Full jitter (Step 3) spreads the retries and is why the fallback uses random.uniform, not a fixed multiplier.
  • Monotonic vs wall-clock time. Use time.monotonic() for the window, never time.time(). An NTP correction or DST shift on wall-clock time can rewind the deque and let a burst through.
  • Unbounded deque memory. At very high RPM the timestamp deque grows with the ceiling. Above ~10,000 RPM, switch to a token-bucket counter that tracks a single float balance instead of every timestamp, capping memory regardless of throughput.
  • Concurrency leaks past the ceiling. RPM limits and concurrency limits are independent. Without the semaphore in Step 2, a slow endpoint lets in-flight calls stack up and blow the connection cap even while RPM looks fine.
  • Co-tenant quota sharing. If several workers poll the same vendor account, the ceiling is shared across all of them. Divide max_requests by the worker count (or centralise the limiter behind one coordinator) — each worker assuming the full ceiling collectively overshoots it.
  • Fallback path for open circuits. When the breaker opens, do not drop the interval — reconcile the gap later via CSV bulk import automation so no transactions are lost while the endpoint recovers.

FAQ

Should I rate limit client-side if the vendor already returns 429?

Yes. Reacting to 429 means you have already wasted a request and risk an IP block after repeated offences. A client-side sliding window keeps you under the ceiling proactively, and the Retry-After handling is the safety net for the rare overshoot — not the primary control.

Sliding window or token bucket — which should I use?

A sliding-window deque is exact and easy to reason about, ideal up to a few thousand RPM. A token bucket uses constant memory and naturally allows short bursts, which suits very high RPM or bursty backfills. Both honour the same effective ceiling; pick the bucket when timestamp memory becomes the constraint.

How do I pick the safety margin?

Start at 0.85 of the documented ceiling and watch your 429 rate. If it stays at zero for a week you can inch toward 0.90; if you see occasional throttling, drop to 0.80. Leave more headroom when several workers share one vendor account, since their traffic sums against the same limit.

Why validate sequence gaps if rate limiting already succeeded?

Rate limiting only guarantees the request was accepted, not that the payload was complete. Vendors under load sometimes return a 200 with a truncated page. The per-store seq diff is an independent completeness check that catches silent truncation before it distorts variance reports.

Where does the limiter sit relative to the polling loop?

It wraps the transport layer only. The delta-sync cursor logic, BOM reconciliation, and cost math live in the parent polling strategy; the limiter is a gate every outbound call passes through, so it composes cleanly with retry and circuit-breaker layers without knowing anything about food-cost semantics.