Python Web Scraping
Reading layout

Replaying Mobile API Requests in Python

A recording is a snapshot; the goal set out in Scraping Mobile App APIs is a client you can run next month without re-capturing anything.

From captured flow to typed rows Five stages run left to right: the raw captured flow, header triage, turning fixed values into parameters, a client class with retries, and typed rows ready to store. Five steps from one recorded call to a client you can run dailyCapturedone flowraw JSONLTriagedrop noisekeep 5 headersParameterspath + querycursor, limitClient classhttpx + retryone sessionTyped rowsdataclassesready to store
A capture is a snapshot; a client is a parameterised, retrying, typed version of that snapshot — each stage removes something that would otherwise rot within a release or two.

Most of the work is subtraction. A captured request carries roughly twenty headers, of which four or five are load-bearing; a token that stopped working before you finished reading it; and a URL with one cursor value frozen into the query string. Delete the headers your HTTP library sets better than you can, replace the token with something that refreshes itself, turn the frozen cursor into a loop variable, and parse the result into typed objects so a renamed field fails loudly instead of quietly.

Which Headers Actually Matter

Three buckets for captured headers Headers that must always be reproduced, headers to test one at a time, and headers that the HTTP client should set itself and that must never be copied from a capture. Sorting the headers in a captured requestAlways reproduceAuthorizationX-App-VersionUser-Agent of the appAcceptthe API reads theseTest one at a timeX-Device-IdAccept-LanguageX-Request-SignatureX-Session-Tokendrop one, then retestNever copy acrossHostContent-LengthConnectionAccept-Encodingthe client sets these
Sort every captured header into one of three buckets before writing any code — the middle bucket is the only one that needs experiment, and the right-hand one causes bugs if copied.

Sort every header in the capture into three buckets before writing code.

Always reproduce. Authorization obviously. The app's own User-Agent, which usually looks like ExampleApp/8.4.1 (Android 14; Pixel 7) and sometimes decides which response schema you get. Any X-App-Version or platform header, because backends use them to route between API versions and to reject builds they no longer support. And Accept, which on some endpoints selects between a compact and a verbose serialisation.

Test one at a time. Device identifiers, Accept-Language, session hints, signature headers. These may be required, may be ignored, and may actively harm you if you send a stale value. The method is mechanical: start from a request that works, remove one header, re-run, and record the result. Ten minutes of this produces a header set you understand, which is worth far more than a header set that works for reasons you cannot name.

Never copy across. Host, Content-Length, Connection, Accept-Encoding and Transfer-Encoding are connection-level. httpx computes all of them from the request it is actually making, and a copied value will disagree with reality — a stale Content-Length truncates or stalls the body, and an Accept-Encoding naming a codec the client cannot decode produces bytes that look like corruption.

CONNECTION_HEADERS = {
    "host",
    "content-length",
    "connection",
    "accept-encoding",
    "transfer-encoding",
    "te",
    "upgrade",
    "proxy-connection",
}


def clean_headers(captured: dict[str, str]) -> dict[str, str]:
    """Drop connection-level headers that the HTTP client must compute itself."""
    return {
        name: value
        for name, value in captured.items()
        if name.lower() not in CONNECTION_HEADERS
    }


if __name__ == "__main__":
    raw = {
        "Host": "api.example.com",
        "User-Agent": "ExampleApp/8.4.1 (Android 14; Pixel 7)",
        "Authorization": "Bearer example-token",
        "Accept-Encoding": "gzip",
        "Content-Length": "0",
        "X-App-Version": "8.4.1",
    }
    print(clean_headers(raw))

The Cookie header deserves a note of its own, because it straddles the buckets. Some mobile clients keep a session cookie alongside the bearer token, and a value pasted from a capture will be both stale and misleading — requests appear to work during testing and start failing hours later for reasons that look like a token problem. Let httpx manage the cookie jar from the responses it receives, and only set a cookie by hand when removing it demonstrably breaks the call.

One more consideration sits underneath the header set: header order. Native HTTP stacks emit a fixed order, and a small number of backends check it. httpx preserves the insertion order of the dictionary you pass, so if order turns out to matter, build the dictionary in the order the capture shows and the problem is solved without any extra library.

Handling Tokens That Expire

Mobile auth is nearly always a short-lived access token plus a long-lived refresh token. Access tokens of five to sixty minutes are typical, which means any crawl worth running will outlive its first token. Hard-coding the captured value produces a client that works for one test and returns 401 forever afterwards.

The fix is to treat the token as state owned by the client rather than a constant. Fetch it lazily, cache it with its expiry, and refresh it when a request comes back unauthorised — once, so a genuinely revoked credential fails fast instead of looping.

import time
from dataclasses import dataclass, field

import httpx


@dataclass
class TokenStore:
    """Holds an access token and knows when it needs replacing."""

    refresh_token: str
    token_url: str
    user_agent: str
    _access_token: str | None = field(default=None, repr=False)
    _expires_at: float = 0.0

    def _refresh(self) -> None:
        headers = {"User-Agent": self.user_agent, "Accept": "application/json"}
        with httpx.Client(timeout=15.0, headers=headers) as client:
            response = client.post(
                self.token_url,
                json={"grant_type": "refresh_token", "refresh_token": self.refresh_token},
            )
            response.raise_for_status()
            payload = response.json()
        self._access_token = payload["access_token"]
        # Renew 60 seconds early so a long request never straddles the boundary.
        self._expires_at = time.monotonic() + float(payload.get("expires_in", 3600)) - 60

    @property
    def access_token(self) -> str:
        if self._access_token is None or time.monotonic() >= self._expires_at:
            self._refresh()
        assert self._access_token is not None
        return self._access_token

    def invalidate(self) -> None:
        self._expires_at = 0.0

The sixty-second margin matters more than it looks. Without it, a token that expires between the moment you read it and the moment the server validates it produces an intermittent 401 that is painful to reproduce.

Signed requests are the case this pattern does not solve. If the capture shows a header such as X-Signature whose value changes for every request, the app is computing an HMAC over the path, body and a timestamp using a key inside the binary. That key is deliberately not available to you, and reproducing it is outside what this material covers — treat it as a sign to look for the vendor's documented interface instead.

Pagination Shapes and Their Stop Conditions

Three pagination shapes and their stop conditions Cursor pagination stops when the cursor field is null, page pagination stops on an empty item list, and offset pagination stops when fewer rows come back than the limit requested. Pagination shapes and when to stop askingcursorpageoffsetnext_cursor comes back in the bodystop when the field is null; never build a cursor value yourselfpage=1, page=2 in the query stringstop on an empty item list, not on a total count you were toldoffset=40 and limit=20 togetherstop when fewer rows arrive than the limit you requested
Each pagination shape has exactly one correct stop condition — using the wrong one either truncates the crawl or loops forever on the last page.

Mobile APIs use three shapes, and each has exactly one correct termination test.

Cursor. The response carries next_cursor, next_page_token or similar, and you send it back verbatim. The value is an opaque server-side bookmark — base64 of a key and a direction, usually — so never construct or decode one. Stop when the field is absent or null. Cursors are the most reliable shape under concurrent writes, because a record inserted mid-crawl cannot shift your position.

Page number. ?page=1, ?page=2. Stop when the returned item list is empty, not when you reach a total_pages value: totals are computed per request and drift as the underlying data changes, and trusting one either truncates the crawl or walks off the end. Page numbers also skip and duplicate records when rows are inserted during a long run.

Offset and limit. ?offset=40&limit=20. Stop when fewer rows come back than the limit you asked for. Watch for a server-side maximum — request 500 and be silently given 100, and an offset loop that adds 500 each time will skip four fifths of the data. Always advance by the number of rows actually received.

def next_offset(current_offset: int, received: int, limit: int) -> int | None:
    """Advance by what the server returned, not by what was requested."""
    if received == 0 or received < limit:
        return None
    return current_offset + received

A Client Class You Can Run on a Schedule

Putting it together: one session, retries on the failures worth retrying, a token that renews itself, cursor pagination, and typed rows. tenacity handles backoff so the retry policy stays declarative rather than tangled into the request method.

from __future__ import annotations

# TokenStore is the class defined in the previous section; keep both in one module.
from collections.abc import Iterator
from dataclasses import dataclass

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

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


class TransientAPIError(RuntimeError):
    """Raised for responses that are worth trying again."""


@dataclass(frozen=True, slots=True)
class CatalogItem:
    item_id: str
    name: str
    price_amount: float
    price_currency: str

    @classmethod
    def from_json(cls, row: dict) -> "CatalogItem":
        price = row.get("price") or {}
        return cls(
            item_id=str(row["id"]),
            name=str(row.get("name", "")).strip(),
            price_amount=float(price.get("amount", 0.0)),
            price_currency=str(price.get("currency", "USD")),
        )


class MobileCatalogClient:
    """A small, durable client for one versioned mobile endpoint."""

    def __init__(self, base_url: str, tokens: TokenStore, app_version: str = "8.4.1") -> None:
        self._tokens = tokens
        self._client = httpx.Client(
            base_url=base_url,
            http2=True,
            timeout=httpx.Timeout(20.0, connect=10.0),
            headers={
                "User-Agent": f"ExampleApp/{app_version} (Android 14; Pixel 7)",
                "X-App-Version": app_version,
                "X-Platform": "android",
                "Accept": "application/json",
            },
        )

    def __enter__(self) -> "MobileCatalogClient":
        return self

    def __exit__(self, *exc_info: object) -> None:
        self._client.close()

    @retry(
        retry=retry_if_exception_type((TransientAPIError, httpx.TransportError)),
        wait=wait_exponential(multiplier=1, min=2, max=30),
        stop=stop_after_attempt(5),
        reraise=True,
    )
    def _get(self, path: str, params: dict[str, object]) -> dict:
        headers = {"Authorization": f"Bearer {self._tokens.access_token}"}
        response = self._client.get(path, params=params, headers=headers)

        if response.status_code == 401:
            self._tokens.invalidate()
            raise TransientAPIError("access token rejected, refreshing")
        if response.status_code in RETRYABLE_STATUS:
            raise TransientAPIError(f"transient status {response.status_code}")

        response.raise_for_status()
        return response.json()

    def iter_items(self, page_size: int = 40) -> Iterator[CatalogItem]:
        cursor: str | None = None
        while True:
            params: dict[str, object] = {"limit": page_size}
            if cursor is not None:
                params["cursor"] = cursor
            payload = self._get("/catalog/items", params)

            rows = payload.get("items", [])
            if not rows:
                return
            for row in rows:
                yield CatalogItem.from_json(row)

            cursor = payload.get("next_cursor")
            if not cursor:
                return


if __name__ == "__main__":
    store = TokenStore(
        refresh_token="a-refresh-token-you-obtained-yourself",
        token_url="https://api.example.com/v3/auth/refresh",
        user_agent="ExampleApp/8.4.1 (Android 14; Pixel 7)",
    )
    with MobileCatalogClient("https://api.example.com/v3", store) as api:
        for index, item in enumerate(api.iter_items(), start=1):
            print(f"{index:>5}  {item.item_id:<12} {item.price_amount:>9.2f} {item.name}")
            if index >= 200:
                break

Four decisions in that class are the ones that keep it alive. A 401 is turned into a retryable error after invalidating the token, so the next attempt refreshes rather than failing. Retries cover only transient statuses and transport errors — a 404 or 422 is a bug in your code and should surface immediately. iter_items is a generator, so memory stays flat regardless of how many pages exist. And from_json centralises the response shape in one place, which is where the failure will appear when the vendor changes something.

Deeper backoff strategy, including jitter and per-host budgets, is covered in Retrying Failed Requests with Tenacity; the nested-JSON reshaping that mobile payloads tend to need is in Parsing JSON and XML Responses.

Edge Cases and Caveats

  • float(price["amount"]) breaks on money strings. Some APIs send "1,299.00" or minor units as integers. Decide the convention from real data, and use Decimal when the values feed anything financial.
  • HTTP/2 is not always a win. http2=True multiplexes well against modern hosts but adds a handshake step; if a host negotiates down to HTTP/1.1 anyway, the flag costs a little and gains nothing.
  • Retrying a non-idempotent POST can duplicate records. The retry decorator above is safe for GET. Wrap writes only when the endpoint documents an idempotency key.
  • A refresh token can be single-use. Some services rotate it on every refresh and invalidate the old one. Persist the new value from the refresh response, or the second run of the day fails.
  • 429 may carry Retry-After. Exponential backoff ignores it. Reading that header and sleeping for exactly the stated interval is both politer and faster than guessing.
  • Empty page versus null cursor. Some endpoints return an empty item list while still handing back a cursor. Checking both conditions, as iter_items does, avoids an infinite loop on that shape.
  • Locale changes the payload. An Accept-Language value carried over from a capture can return translated names and localised prices. Set it explicitly rather than inheriting it by accident.

Frequently Asked Questions

How do I know which headers I can safely drop? Empirically, one at a time. Start from a request that succeeds, remove a single header, re-run, and note whether the response changes. Most captures reduce to four or five meaningful headers, and knowing which ones they are makes every future failure faster to diagnose.

What happens when the token expires mid-crawl? With the pattern above, nothing visible: the 401 invalidates the cached token, the retry refreshes it, and the same page is requested again with the new credential. The cursor is held in the generator's local state, so no position is lost.

Should I use requests instead of httpx? Either works. httpx is used here for HTTP/2, native timeouts per phase, and an async path you can move to later without rewriting the parsing. If a project already standardises on requests, the same structure — one session, one token store, one generator — applies unchanged.

How do I keep the client working after an app update? Save a real response body as a fixture and assert its shape in a test. Version bumps in the path and renamed fields are the two things that break these clients, and both show up as a failing assertion in seconds rather than as a table full of nulls a week later.

Back to home