Understand the rate-limit response

A 429 response usually indicates that a request rate or quota has been exceeded. Repeating the request immediately can increase pressure and prolong the incident. Determine whether the limit applies to an account, network address or shared service before changing client behavior. This tutorial implements a small read-only client with explicit retry and waiting budgets, making its decisions inspectable rather than hiding them inside an unlimited background loop.

Retry-After can contain a number of seconds or an HTTP date. The parser handles both forms and treats a past date as zero remaining delay. When the requested delay exceeds the client's accepted waiting budget, the operation stops. Retrying earlier would not respect the server's instruction. An absent or invalid header uses bounded randomized backoff, which spreads attempts without promising that a particular provider will immediately accept them.

Define waiting and retry budgets

The example permits one to five attempts and gives each network request its own timeout. Those controls are not the same as an overall deadline for a larger job. A production design also needs to account for total duration, connection counts and quotas shared by multiple workers. Do not copy a retry policy for a read-only GET into payments or order creation without a separate contract preventing repeated side effects.

Save the four complete files together and run python -m unittest test_retry_client.py. Tests cover both header formats and an actual local server returning 429 before succeeding. The test records requested sleeps instead of waiting in real time, keeping the experiment fast and reproducible. To observe the actual delay, run python demo_server.py in one terminal and python demo_client.py in another terminal on the same test machine.

The demonstration server requests a one-second wait on its first call and then returns a successful ready value. This verifies the local policy, not a real provider's quota implementation. Inspect the provider's documentation, account state, real response headers and timestamps before adapting the settings. Keep access keys and private request data out of public diagnostic reports; useful timing and status information should not require publishing the complete sensitive payload.

Complete executable files

The four files are complete. The article provides prerequisites, command order and verification; code is neither shortened nor hidden.

1. Complete retry_client.py

Save this entire file beside the others using the exact filename.

import email.utils
import json
import random
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone


def retry_delay(value, now=None):
    if not value:
        return None
    value = value.strip()
    if value.isascii() and value.isdigit():
        return float(value)
    try:
        target = email.utils.parsedate_to_datetime(value)
        if target.tzinfo is None:
            return None
        return max(0.0, (target - (now or datetime.now(timezone.utc))).total_seconds())
    except (TypeError, ValueError, OverflowError):
        return None


def fetch_json(url, attempts=3, max_wait=30, sleep=time.sleep):
    if not 1 <= attempts <= 5 or max_wait < 0:
        raise ValueError('Invalid retry budget')
    for attempt in range(attempts):
        try:
            with urllib.request.urlopen(url, timeout=5) as response:
                body = response.read(1_000_001)
                if len(body) > 1_000_000:
                    raise ValueError('Response exceeds byte limit')
                return json.loads(body)
        except urllib.error.HTTPError as error:
            if error.code != 429 or attempt == attempts - 1:
                raise
            delay = retry_delay(error.headers.get('Retry-After'))
            if delay is None:
                delay = random.uniform(0, min(max_wait, 2 ** attempt))
            if delay > max_wait:
                raise  # Do not retry earlier than the server requested.
            error.close()
            sleep(delay)

The execution section explains this file’s command, expected outcome and limits.

2. Complete demo_server.py

Save this entire file beside the others using the exact filename.

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


class Handler(BaseHTTPRequestHandler):
    requests = 0

    def do_GET(self):
        type(self).requests += 1
        limited = type(self).requests == 1
        body = b'{"ready":true}'
        self.send_response(429 if limited else 200)
        if limited:
            self.send_header('Retry-After', '1')
        self.send_header('Content-Type', 'application/json')
        self.send_header('Content-Length', str(len(body)))
        self.end_headers()
        self.wfile.write(body)


if __name__ == '__main__':
    ThreadingHTTPServer(('127.0.0.1', 8087), Handler).serve_forever()

The execution section explains this file’s command, expected outcome and limits.

3. Complete demo_client.py

Save this entire file beside the others using the exact filename.

import json
from retry_client import fetch_json

if __name__ == '__main__':
    print(json.dumps(fetch_json('http://127.0.0.1:8087/')))

The execution section explains this file’s command, expected outcome and limits.

4. Complete test_retry_client.py

Save this entire file beside the others using the exact filename.

import threading
import unittest
from datetime import datetime, timezone
from http.server import ThreadingHTTPServer
from demo_server import Handler
from retry_client import fetch_json, retry_delay


class RetryTests(unittest.TestCase):
    def test_header_formats(self):
        now = datetime(2026, 10, 27, tzinfo=timezone.utc)
        self.assertEqual(retry_delay('12', now), 12)
        self.assertEqual(retry_delay('Tue, 27 Oct 2026 00:00:05 GMT', now), 5)
        self.assertIsNone(retry_delay('invalid', now))

    def test_real_429_then_success(self):
        Handler.requests = 0
        server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        sleeps = []
        try:
            self.assertTrue(fetch_json(f'http://127.0.0.1:{server.server_port}/', sleep=sleeps.append)['ready'])
            self.assertEqual(sleeps, [1])
            self.assertEqual(Handler.requests, 2)
        finally:
            server.shutdown(); server.server_close(); thread.join()


if __name__ == '__main__':
    unittest.main()

The execution section explains this file’s command, expected outcome and limits.

Run the example and evaluate the outcome

If rate limiting persists, increasing the retry count is rarely the first useful change. Coalesce equivalent requests, cache appropriate results and slow the producer to match destination capacity. Evaluate the reduction in unnecessary requests and the ability to stop with an actionable error in bounded time. These files provide a reproducible starting point, while quota coordination, monitoring and behavior across independent processes remain application-level responsibilities.

Implementation checklist

  • Test both Retry-After formats.
  • Keep attempt and waiting budgets explicit.
  • Assess mutating operations separately.
  • Do not hide permanent failures with unlimited retries.

References

Editorial guidance: Liyan Knowledge.

Published: