Build a Resilient REST API Poller with Python

Admin

October 4, 2026

8 min read

An automation job should not hang forever when an API is slow, and a temporary service error should not immediately discard the whole run. This example uses Python Requests to poll a read-only inventory endpoint, set connection and response timeouts, retry selected transient responses, and follow cursor-based pages.

Replace the example URL and JSON field names with the API’s documented schema. The script requires an HTTPS base URL before it adds the bearer token to requests. Use a read-only token with the narrowest available scope and keep it in an environment variable or approved secret store.

Step 1: Install Requests and Set the API Variables

Prepare the Environment and Read-Only Token

Setup

Install Requests in a virtual environment. Set the token in your shell for the current session; avoid typing it into a script or command line where it can be retained in history. The hostname below is an example and must be replaced with your service endpoint.

python3 -m venv .venv
source .venv/bin/activate
python -m pip install requests
export INVENTORY_API_BASE="https://api.example.internal/v1"
read -rsp "Read-only API token: " INVENTORY_API_TOKEN
export INVENTORY_API_TOKEN
printf '\n'

Step 2: Configure a Session with Timeouts and Bounded Retries

Retry Safe GET Requests on Transient Errors

Reliability

A Requests session reuses connections. This adapter retries only GET requests for selected temporary responses, honors a server’s Retry-After header, and applies a backoff between attempts. Redirects are rejected so the bearer token is sent only to the configured HTTPS endpoint. The per-request timeout bounds connection setup and waiting for response data; retries are deliberately limited.

import os
from urllib.parse import urlsplit

import requests
from requests.adapters import HTTPAdapter
from urllib3.util import Retry

base_url = os.environ["INVENTORY_API_BASE"].rstrip("/")
token = os.environ["INVENTORY_API_TOKEN"]
parsed_base = urlsplit(base_url)
if (
    parsed_base.scheme != "https"
    or not parsed_base.hostname
    or parsed_base.username is not None
    or parsed_base.password is not None
    or parsed_base.query
    or parsed_base.fragment
):
    raise ValueError("INVENTORY_API_BASE must be an HTTPS URL without credentials, query, or fragment")

retry_policy = Retry(
    total=4,
    connect=3,
    read=2,
    status=3,
    backoff_factor=0.5,
    status_forcelist=(429, 502, 503, 504),
    allowed_methods=frozenset({"GET"}),
    respect_retry_after_header=True,
)

session = requests.Session()
session.headers.update({
    "Authorization": f"Bearer {token}",
    "Accept": "application/json",
})
session.mount("https://", HTTPAdapter(max_retries=retry_policy))

Step 3: Follow Cursor-Based Pagination

Fetch Each Page Until the API Has No Cursor

Pagination

Many APIs return a page of items and a cursor for the next page. This example expects an object with an items array and optional next_cursor. Check the API documentation for its actual response shape and pagination limits. Repeated cursors and an explicit maximum page count both stop a malformed or unexpectedly long traversal.

items = []
cursor = None
seen_cursors = set()
MAX_PAGES = 1000

for _ in range(MAX_PAGES):
    params = {"limit": 100}
    if cursor:
        params["cursor"] = cursor

    response = session.get(
        f"{base_url}/assets",
        params=params,
        timeout=(3.1, 20),
        allow_redirects=False,
    )
    if 300 <= response.status_code < 400:
        raise RuntimeError(
            "API redirected the request; configure its final HTTPS endpoint directly"
        )
    response.raise_for_status()
    page = response.json()
    if not isinstance(page, dict) or not isinstance(page.get("items"), list):
        raise ValueError("API response must contain an items array")
    items.extend(page.get("items", []))
    next_cursor = page.get("next_cursor")
    if not next_cursor:
        break
    if not isinstance(next_cursor, str):
        raise ValueError("API next_cursor must be a string or null")
    if next_cursor in seen_cursors:
        raise RuntimeError("API repeated a pagination cursor")
    seen_cursors.add(next_cursor)
    cursor = next_cursor
else:
    raise RuntimeError(f"API exceeded the {MAX_PAGES}-page safety limit")

print(f"Retrieved {len(items)} asset records")

Step 4: Save a Minimal Report and Close the Session

Export Only the Fields You Need

Reporting

Select the fields needed by the operator who will review the report. JSON preserves nested API data, while CSV is easier to open in common tools. Treat the output as sensitive inventory, store it in an access-controlled location, and close the session when finished.

import json
from pathlib import Path

report = [
    {
        "id": item.get("id"),
        "name": item.get("name"),
        "status": item.get("status"),
    }
    for item in items
]
Path("asset-report.json").write_text(
    json.dumps(report, indent=2), encoding="utf-8"
)
session.close()

See the Requests advanced usage guide for sessions, adapters, retries, and timeout behavior.