Building a Reusable Ingestion Client

The previous three articles in this section each introduced one concern in isolation: a shared session and timeouts, pagination and backoff, and token caching. A real integration needs all three at once, and a second integration with a different API needs the same three again. Repeating them by hand per source is exactly the situation Don't repeat yourself warns about: the retry logic is one piece of knowledge, and it should have one authoritative implementation, not five copies drifting slowly apart as each gets patched independently.

One base class, source-specific subclasses#

The facade pattern — one simplified interface in front of several moving parts — is the right shape here. A base class owns the session, the timeout, the retry loop, and the pagination walk; a subclass per source only supplies what is actually different: the base URL, the auth header, and the shape of one page of results. The required override below is spelled out by hand with NotImplementedError rather than Python's abc module, since one required method does not yet justify the extra import — abc.abstractmethod earns its place once a base class grows several required overrides.

The constructor takes only what every source needs — a base URL and a timeout — and delegates the one thing that differs per source to a method it does not implement itself:

class BaseAPIClient:    def __init__(self, base_url: str, timeout: tuple[float, float] = (3.05, 10)):        self._base_url = base_url.rstrip("/")        self._timeout = timeout        self._session = requests.Session()        self._session.headers.update(self._auth_headers())
    def _auth_headers(self) -> dict:        """Override in a subclass: static API key, Basic Auth, or a cached OAuth token."""        raise NotImplementedError

Calling self._auth_headers() from __init__ before any subclass has defined it looks backwards, but it is the normal shape for this kind of hook: by the time __init__ actually runs, Python has already resolved self to the subclass instance, so self._auth_headers() calls the subclass's version, not this stub. The base class only supplies the fallback that fires if a subclass forgets to override it at all.

The retry decision — which failures are worth retrying — is one clause, small enough to name and test on its own rather than bury inside a longer function:

def _is_retryable(status: int) -> bool:    return status >= 500 or status == 429

This is the same status < 500 and status != 429 check from Pagination and Rate Limiting, just given a name. Naming it turns "which errors get retried" into something the rest of the class can read at a glance, instead of a condition a reader has to re-derive from an if buried inside a loop.

_request_with_backoff is the loop that uses it — attempt the request, and on a retryable failure, wait and try again:

def _request_with_backoff(self, method: str, path: str, max_attempts=5, **kwargs):    url = f"{self._base_url}{path}"    for attempt in range(max_attempts):        try:            response = self._session.request(method, url, timeout=self._timeout, **kwargs)            response.raise_for_status()            return response        except requests.HTTPError as exc:            status = exc.response.status_code            if not _is_retryable(status) or attempt == max_attempts - 1:                raise            time.sleep(2 ** attempt + random.uniform(0, 1))

This is the exponential-backoff-with-jitter loop from the same article, pulled into the base class rather than copied per call site — every subclass gets it for free instead of re-implementing it.

Pagination is the last piece: walk the cursor from Pagination and Rate Limiting, generalized to yield results lazily instead of building a full list, so a caller that only needs the first few results can stop early without paying for pages it never asked for:

def paginate(self, path: str, params: dict | None = None):    """Yield every result across every page, following each page's next cursor."""    params = dict(params or {})    cursor = None    while True:        if cursor:            params["cursor"] = cursor        page = self._request_with_backoff("GET", path, params=params).json()        yield from page["results"]        cursor = page.get("next_cursor")        if cursor is None:            break

Three methods, each doing one job — authenticate, retry, paginate — is the whole base class. import time, import random, and import requests sit at the top of the module as usual and are omitted from the snippets above.

A source-specific subclass#

class OrdersAPIClient(BaseAPIClient):    def __init__(self, base_url: str, api_key: str):        self._api_key = api_key        super().__init__(base_url)
    def _auth_headers(self) -> dict:        return {"Authorization": f"Bearer {self._api_key}", "Accept": "application/json"}
    def get_orders(self, **params) -> list[dict]:        return list(self.paginate("/orders", params))

A second source with OAuth 2.0 client credentials instead of a static key overrides only _auth_headers, reusing everything else:

class BillingAPIClient(BaseAPIClient):    def __init__(self, base_url: str, token_url: str, client_id: str, client_secret: str):        self._token_cache = OAuthTokenCache(token_url, client_id, client_secret)        super().__init__(base_url)
    def _auth_headers(self) -> dict:        return {"Authorization": f"Bearer {self._token_cache.get_token()}"}
    def get_invoices(self, **params) -> list[dict]:        return list(self.paginate("/invoices", params))

OAuthTokenCache is the class from API Authentication Patterns, reused as-is. Neither subclass re-implements the retry loop, the pagination walk, or the timeout — the only code specific to each source is the base URL, how it authenticates, and the one method name that makes sense for its domain.

Where this stops paying off#

A base class earns its keep when at least two real sources share the concern it centralizes. It stops paying off past a certain point:

• Different pagination shapes. A source using page numbers instead of cursors, or nesting results one level deeper, needs its own paginate override — forcing it through the shared one just to avoid writing a second method usually produces worse code than the duplication would have.
• Business logic creeping into the client. get_orders should return raw API results, not filtered or transformed ones. Keeping the single-responsibility principle in mind here means the client's one job is talking to the API — parsing business meaning out of the response belongs in the calling code, not the client.
• A deep subclass hierarchy. One base class with source-specific subclasses is enough for this problem. If a second layer of subclassing starts to feel necessary — a subclass of a subclass to share one more override — that is usually a sign to switch to composition over inheritance: pass the retry policy and the pagination strategy in as objects instead of encoding them in the class hierarchy.

Common mistakes#

Building the abstraction before there is a second source#

One integration does not reveal what actually varies between sources versus what is incidental to this particular API. Write the concrete client first; generalize once a second source shows what the base class actually needs to own.

Putting transformation logic inside the client#

A client method that filters, renames, or reshapes fields mixes transport concerns with business logic, making both harder to test independently. Return the API's data as-is; transform it in the caller.

Swallowing exceptions inside _request_with_backoff#

The retry loop should retry retryable failures and re-raise everything else — it should never catch an exception and return None silently. A caller that cannot tell "no results" from "the request failed" cannot handle either case correctly.

Further reading#

• Don't repeat yourself — the principle this whole article is an applied example of.
• Facade pattern — the design pattern the base class follows.
• Single-responsibility principle — why business logic does not belong inside the client.
• Composition over inheritance — the alternative worth reaching for once a subclass hierarchy grows past one level.
• abc — Abstract Base Classes — the standard-library tool for enforcing required overrides formally.
• Connecting to PostgreSQL from Python — the same "one shared client" idea applied to a database driver.

See also#