رفع خطای ۴۲۹؛ مدیریت محدودیت درخواست‌ها

این فایل را با همین نام و بدون حذف سطرها کنار بقیهٔ فایل‌ها ذخیره کنید.

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)

دستور اجرا، نتیجهٔ مورد انتظار و محدودیت این فایل در بخش اجرای مقاله توضیح داده شده است.

این فایل را با همین نام و بدون حذف سطرها کنار بقیهٔ فایل‌ها ذخیره کنید.

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()

دستور اجرا، نتیجهٔ مورد انتظار و محدودیت این فایل در بخش اجرای مقاله توضیح داده شده است.

این فایل را با همین نام و بدون حذف سطرها کنار بقیهٔ فایل‌ها ذخیره کنید.

import json
from retry_client import fetch_json

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

دستور اجرا، نتیجهٔ مورد انتظار و محدودیت این فایل در بخش اجرای مقاله توضیح داده شده است.

این فایل را با همین نام و بدون حذف سطرها کنار بقیهٔ فایل‌ها ذخیره کنید.

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()

دستور اجرا، نتیجهٔ مورد انتظار و محدودیت این فایل در بخش اجرای مقاله توضیح داده شده است.