Add bounded retries and a working collector

A timeout is an uncertain transport outcome

A timeout tells the client that it did not obtain the expected response within its transport boundary. For a read-only operation, a bounded retry may be reasonable. For a mutation, a timeout does not prove the server did nothing. The server could have created a ticket while the response was lost.

Our fictional collection operation is a GET-style read. The client retries a timeout at most twice after the first failed attempt. A network adapter must actually honor the supplied timeout. A timeout argument in a function signature alone does not enforce a real deadline; the in-memory fixture simply records it and raises the chosen fault.

The example does not implement a complete wall-clock deadline for the whole collection. Default page and attempt budgets bound counts, while each real request would need an effective timeout. A production integration may need a total execution deadline, cancellation and scheduler behavior. Those are explicit design tasks in the project section.

Respect rate limits within a delay budget

For a retryable status, the example uses a small backoff unless it receives an accepted Retry-After delay. It supports non-negative decimal delay seconds. HTTP also permits a date form; this first adapter does not interpret that form and falls back to its bounded backoff. A real provider adapter must document and test its date handling rather than claim full header support.

If a supported delay is longer than the client's five-second wait budget, the code stops and reports that limitation. It does not shorten a requested thirty-second wait to five seconds and then hammer the provider too soon. A scheduler could defer the job for the requested interval; that is a different policy from waiting inside this small collector.

The offline examples record delays in a list. They do not sleep or contact a server. This lets you verify retry behavior immediately and deterministically. It also means that a passing lab does not validate real timing, network cancellation or provider rate-limit semantics.

Keep mutation retries separate

An operation is idempotent when repeating the same request has the same intended effect as making it once under its contract. That does not mean every response body or log entry must be identical. GET is intended to be safe; PUT and DELETE have idempotent semantics in HTTP, but an API's implementation and business workflow still need verification. A generic POST retry is not automatically safe.

For a ticket-creation endpoint, a provider-supported idempotency key or a durable request identity may let the client recover from an ambiguous response. Merely generating a random header does not make the server honor it. The provider must define key scope, retention, conflict behavior and a way to establish the prior result.

This course implements no mutation path. Its collector cannot create, delete or reassign an asset. In your project design, explicitly keep the read transport and any future write workflow separate so a retry helper meant for reads is not casually reused for state-changing operations.

Save the complete example

Save this original course program as asset_client.py. FixtureTransport is the whole transport for the lab: it returns selected responses and records calls. There is no hidden network implementation.

"""Offline GET-style page collector. It has no network adapter or credentials."""
from dataclasses import dataclass
import json
import re
from urllib.parse import parse_qs, urlsplit

BASE = 'https://api.example.invalid/v1/assets'
MAX_BODY = 500_000
MAX_ITEMS = 1_000

class ApiError(Exception):
    pass

@dataclass
class Response:
    status: int
    headers: dict
    body: bytes


def cursor_key(url):
    if not isinstance(url, str):
        raise ApiError('Page URL must be text')
    try:
        parsed = urlsplit(url)
    except ValueError as error:
        raise ApiError('Malformed page URL') from error
    if parsed.scheme != 'https' or parsed.netloc != 'api.example.invalid' or parsed.path != '/v1/assets' or parsed.fragment:
        raise ApiError('Page URL is outside the approved origin/resource')
    params = parse_qs(parsed.query, keep_blank_values=True)
    if not params:
        return ''
    if set(params) != {'cursor'} or len(params['cursor']) != 1 or not re.fullmatch(r'[A-Za-z0-9_-]{1,64}', params['cursor'][0]):
        raise ApiError('Invalid pagination cursor')
    return params['cursor'][0]


def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ApiError('Duplicate JSON property')
        result[key] = value
    return result


def reject_constant(value):
    raise ApiError('Non-standard JSON numeric constant')


def parse_page(response):
    if not isinstance(response.body, bytes) or len(response.body) > MAX_BODY:
        raise ApiError('Response body exceeds the accepted contract')
    raw_type = response.headers.get('content-type', '')
    if not isinstance(raw_type, str):
        raise ApiError('Content-Type must be text')
    content_type = raw_type.split(';', 1)[0].strip().lower()
    if content_type != 'application/json':
        raise ApiError('Expected application/json')
    try:
        page = json.loads(response.body.decode('utf-8'), object_pairs_hook=unique_object, parse_constant=reject_constant)
    except (ValueError, UnicodeError) as error:
        raise ApiError('Invalid JSON response') from error
    if not isinstance(page, dict) or set(page) != {'items', 'next', 'total'}:
        raise ApiError('Unexpected page schema')
    if type(page['total']) is not int or not 0 <= page['total'] <= MAX_ITEMS:
        raise ApiError('Invalid advertised total')
    if not isinstance(page['items'], list):
        raise ApiError('items must be an array')
    if page['next'] is not None:
        cursor_key(page['next'])
    for row in page['items']:
        if not isinstance(row, dict) or set(row) != {'id', 'owner_id', 'site'}:
            raise ApiError('Unexpected item schema')
        if not isinstance(row['id'], str) or not re.fullmatch(r'A-[0-9]{3}', row['id']):
            raise ApiError('Invalid item identity')
        if not isinstance(row['owner_id'], str) or not re.fullmatch(r'u[0-9]{3}', row['owner_id']):
            raise ApiError('Invalid owner identity')
        if row['site'] not in ('HQ', 'BRANCH'):
            raise ApiError('Invalid site code')
    return page


def read_page(transport, url, sleep, attempts=3):
    for attempt in range(attempts):
        response = None
        try:
            response = transport.fetch(url, timeout=5)
        except TimeoutError:
            retryable = True
        else:
            if type(response.status) is not int or not 200 <= response.status <= 599:
                raise ApiError('Invalid HTTP status contract')
            if response.status == 200:
                return parse_page(response)
            retryable = response.status in (429, 503)
            if not retryable:
                raise ApiError(f'Non-retryable HTTP status {response.status}')
        if attempt == attempts - 1:
            raise ApiError('Page request exhausted its retry budget')
        delay = min(5, 2 ** attempt)
        if response is not None:
            raw = response.headers.get('retry-after', '')
            if isinstance(raw, str) and re.fullmatch(r'[0-9]{1,6}', raw):
                seconds = int(raw)
                if seconds > 5:
                    raise ApiError('Retry-After exceeds this client delay budget')
                delay = seconds
        sleep(delay)
    raise ApiError('No page response')


def collect(transport, sleep, max_pages=10):
    if type(max_pages) is not int or not 1 <= max_pages <= 100:
        raise ApiError('max_pages must be an integer from 1 to 100')
    url = BASE
    visited, identities = set(), set()
    items, total = [], None
    for page_number in range(max_pages):
        cursor = cursor_key(url)
        if cursor in visited:
            raise ApiError('Pagination cycle detected')
        visited.add(cursor)
        page = read_page(transport, url, sleep)
        if total is None:
            total = page['total']
        elif total != page['total']:
            raise ApiError('Advertised total changed across pages')
        for row in page['items']:
            if row['id'] in identities:
                raise ApiError('Duplicate identity across collected pages')
            identities.add(row['id'])
            items.append(row.copy())
        if len(items) > MAX_ITEMS or len(items) > total:
            raise ApiError('Collected population exceeds its advertised limit')
        if page['next'] is None:
            if len(items) != total:
                raise ApiError('Terminal population does not reconcile to advertised total')
            return sorted(items, key=lambda row: row['id'])
        url = page['next']
    raise ApiError('Pagination exceeded the page budget')


class FixtureTransport:
    def __init__(self, responses):
        self.responses = {key: list(value) for key, value in responses.items()}
        self.calls = []

    def fetch(self, url, timeout):
        self.calls.append((url, timeout))
        choices = self.responses.get(url, [])
        if not choices:
            raise ApiError('No response remaining for this fixture URL')
        result = choices.pop(0)
        if isinstance(result, Exception):
            raise result
        return result


def json_response(page, status=200, headers=None):
    values = {'content-type': 'application/json'}
    values.update(headers or {})
    return Response(status, values, json.dumps(page).encode('utf-8'))


def baseline_transport():
    return FixtureTransport({
        BASE: [json_response({'items': [
            {'id':'A-002','owner_id':'u102','site':'HQ'},
            {'id':'A-001','owner_id':'u101','site':'HQ'},
        ], 'next':BASE+'?cursor=page2','total':3})],
        BASE+'?cursor=page2': [json_response({'items':[
            {'id':'A-003','owner_id':'u103','site':'BRANCH'},
        ],'next':None,'total':3})],
    })


if __name__ == '__main__':
    transport = baseline_transport()
    delays = []
    print(json.dumps(collect(transport, delays.append), indent=2))
    print(f'{len(transport.calls)} fixture requests; simulated retry delays: {delays}')

Run python asset_client.py. The printed result must contain three sorted assets and say that two fixture requests occurred with no retry delays. Trace the first collect iteration: validate the initial URL, read and parse page one, record its identities, then move to the approved page-two link. Trace the second iteration and its terminal-count reconciliation.

Notice that the collection function returns only after the terminal population has validated. On an error it raises ApiError rather than returning the partially accumulated list. A caller must not catch that error and treat an empty replacement list as a successful collection. Error handling is part of the integration contract.

Inject a rate limit without waiting

Create a short practice runner beside the program:

from asset_client import BASE, FixtureTransport, Response, collect, json_response

transport = FixtureTransport({
    BASE: [
        Response(429, {'retry-after': '2'}, b''),
        json_response({'items': [], 'next': None, 'total': 0}),
    ]
})
delays = []
print(collect(transport, delays.append))
print(delays)
print(len(transport.calls))

Expected outputs are an empty accepted collection, the delay list [2], and two calls. The empty collection is accepted because the valid terminal page advertises total zero. It does not arise from converting the 429 into an empty result.

Replace 429 with 403. The client should stop after one call and never record a retry delay. Replace the delay with 30. It should stop because that delay exceeds the example's wait budget, rather than retry after two seconds. These cases test a policy distinction, not just whether a loop executes.

Do not hide response faults behind a success message

Try a response with status 200, HTML content type and a sign-in message. The parser must reject it. Try valid JSON with a string total "3", a duplicate JSON key, or a missing item identifier. Those must also fail. A successful transport exchange is not a successful inventory collection.

When connecting this design to a real HTTP library, define how it exposes non-success responses and headers. Normalize header names consistently in the adapter. Disable or explicitly validate redirects before forwarding authentication to any redirected destination. The offline origin check cannot protect an automatic redirect hidden inside a future network library.

Log a safe operation identifier, attempt count and failure category for troubleshooting. Keep access tokens, raw authorization headers and unnecessary asset details out of routine logs. The lab's test history deliberately records URL and timeout only, and the URL contains no credential.

Worked laboratory solution

Compare your reasoning with the worked solution

The outcome is unknown to the client, not definitely failed. Reusing the original key with the same request under the API’s contract allows reconciliation with T-9; a new key describes a different operation and can create a duplicate. Persist the operation identity before sending if recovery across process restarts is required. Do not generalize this behavior to APIs that do not document it. When no idempotency support exists, use an approved query/reconciliation workflow or surface the uncertainty for an operator.

Official reference for further study

Download the course workbook and practice materials