Calling REST APIs with Python

Data Ingestion Fundamentals covers the pull side of ingestion in the abstract. In practice, "pull" from an HTTP source almost always means requests — Python's de facto standard HTTP client. This article covers building a small client around it that is safe to call repeatedly, rather than scattering requests.get(...) calls through a codebase.

A minimal call, and what is missing from it#

import requests
response = requests.get("https://api.example.com/v1/orders")data = response.json()

This works for a one-off script and fails in three specific ways under real conditions: it has no timeout, it does not check whether the call actually succeeded, and it opens a new TCP connection on every call instead of reusing one. Each is worth fixing before this code runs on a schedule.

Reuse a client instead of reconnecting every call#

Connecting to PostgreSQL from Python introduces the same idea for a database driver: build one client object per process and pass it around, rather than paying connection setup on every call. A requests.Session is the HTTP equivalent — it keeps a connection pool open across requests to the same host and lets you set headers and auth once instead of on every call:

import requests
def build_client(base_url: str, api_key: str) -> requests.Session:    session = requests.Session()    session.headers.update({        "Authorization": f"Bearer {api_key}",        "Accept": "application/json",    })    session.base_url = base_url    return session

requests.Session has no built-in concept of a base URL, so a thin wrapper function — or a small subclass — is the usual way to avoid repeating the host in every call site. The pattern is the same one used for the psycopg/SQLAlchemy client in Connecting to PostgreSQL: one object, built once, reused everywhere a call to this API is needed.

Never leave the timeout unset#

requests does not apply a default timeout. Without one, a call that never gets a response from the server — not an error, just silence — hangs the calling process forever:

response = session.get(    f"{base_url}/orders",    timeout=(3.05, 10),)

The tuple form sets a connect timeout (time to establish the TCP connection — 3.05 seconds, chosen because it is slightly above a multiple of 3, a common TCP retransmission window) and a read timeout (time to wait for the server to send bytes once connected — 10 seconds here, but the right value depends on the endpoint). A single float applies the same value to both phases. Either form is acceptable; no timeout at all is not.

Treat non-2xx as an error, explicitly#

requests does not raise an exception for a 404 or a 500 by default — response.json() will happily try to parse an HTML error page as JSON and fail with a confusing message unrelated to the real problem. Call raise_for_status() before touching the body:

response = session.get(f"{base_url}/orders", timeout=(3.05, 10))response.raise_for_status()data = response.json()

raise_for_status() raises requests.HTTPError for any 4xx or 5xx status, with the status code and body attached to the exception — an immediate, specific failure instead of a downstream KeyError from code that assumed data was the shape it expected.

The 4xx/5xx split matters for what you do next: a 4xx (bad request, unauthorized, not found) means the request was wrong and retrying it unchanged will fail again the same way. A 5xx means the server had a problem, and retrying after a short wait is often the right response — Pagination and Rate Limiting covers exactly this distinction when it comes to retry logic.

Wrapping requests-specific exceptions#

Code that calls this client should not need to import requests itself just to catch its exceptions. A small wrapper keeps the rest of the codebase decoupled from the HTTP library underneath it:

class UpstreamAPIError(Exception):    """Raised when the upstream API cannot fulfill a request."""
def get_orders(session: requests.Session, base_url: str) -> list[dict]:    try:        response = session.get(f"{base_url}/orders", timeout=(3.05, 10))        response.raise_for_status()    except requests.RequestException as exc:        raise UpstreamAPIError(f"Failed to fetch orders: {exc}") from exc    return response.json()

requests.RequestException is the base class for every exception the library raises — connection errors, timeouts, and HTTPError all inherit from it — so this one except clause is enough to catch anything that can go wrong in the call. from exc preserves the original traceback for debugging while presenting callers with one exception type they actually need to handle.

Full client example#

Putting the session, the headers, and the exception type together, the constructor is everything from the "reuse a client" section above, moved into a class instead of a free function:

class UpstreamAPIError(Exception):    """Raised when the upstream API cannot fulfill a request."""
class OrdersAPIClient:    def __init__(self, base_url: str, api_key: str):        self._base_url = base_url.rstrip("/")        self._session = requests.Session()        self._session.headers.update({            "Authorization": f"Bearer {api_key}",            "Accept": "application/json",        })

And get_orders is the timeout, raise_for_status, and exception-wrapping from the sections above, now reading self._base_url and self._session instead of taking them as parameters:

    def get_orders(self, **params) -> list[dict]:        try:            response = self._session.get(                f"{self._base_url}/orders", params=params, timeout=(3.05, 10)            )            response.raise_for_status()        except requests.RequestException as exc:            raise UpstreamAPIError(f"Failed to fetch orders: {exc}") from exc        return response.json()

Building the class once at process start and calling get_orders() from every task that needs order data gives every caller the same timeout, the same auth header, and the same error handling — none of it repeated at the call site. Building a Reusable Ingestion Client extends this same shape with retries and pagination built in.

Common mistakes#

Calling requests.get directly, repeatedly, across a codebase#

Every call site ends up with its own copy of the timeout, the headers, and the error handling — or, more commonly, without them, because copying boilerplate correctly to every call site does not happen in practice. One client object, built once, avoids the duplication entirely.

Assuming response.json() always returns the expected shape#

An API can return {"error": "..."} with a 200 status, an empty body, or a list where an object was expected. Validate the shape you actually got, especially for third-party APIs whose contract you do not control.

Retrying every failure identically#

A 401 Unauthorized will not succeed on retry without fixing the credentials first; retrying it in a loop just wastes time and can trigger rate limiting or account lockouts. Match the retry strategy to the failure — covered next in Pagination and Rate Limiting.

Further reading#

• Requests: HTTP for Humans — the full library documentation.
• Session objects — connection reuse, default headers, and other advanced usage.
• List of HTTP status codes — the full 4xx/5xx taxonomy referenced above.
• Representational state transfer — the architectural style most JSON APIs loosely follow.
• Connection pool — the general pattern Session implements for HTTP.

See also#