การจำกัดอัตราและตรรกะการลองใหม่

การจำกัดอัตราคืออะไร

import requests

response = requests.get(
    'https://api.example.com/data',
    headers={'Authorization': 'Bearer YOUR_KEY'}
)

if response.status_code == 429:
    print('Rate limit exceeded!')
    # Check headers for limit details
    limit = response.headers.get('X-RateLimit-Limit')
    remaining = response.headers.get('X-RateLimit-Remaining')
    reset = response.headers.get('X-RateLimit-Reset')
    print(f'Limit: {limit}, Remaining: {remaining}, Reset: {reset}')

ส่วนหัว Retry-After

import requests
import time

def request_with_retry_after(url, headers):
    response = requests.get(url, headers=headers)

    if response.status_code == 429:
        retry_after = int(response.headers.get('Retry-After', 60))
        print(f'Rate limited. Waiting {retry_after} seconds...')
        time.sleep(retry_after)

        # Retry once after waiting
        response = requests.get(url, headers=headers)

    response.raise_for_status()
    return response.json()

การหน่วงเวลาแบบทวีคูณ

import requests
import time

def get_with_exponential_backoff(url, headers, max_retries=5):
    for attempt in range(max_retries):
        response = requests.get(url, headers=headers, timeout=(5, 30))

        if response.status_code == 200:
            return response.json()

        if response.status_code in (429, 500, 502, 503):
            wait = 2 ** attempt  # 1, 2, 4, 8, 16 seconds
            print(f'Attempt {attempt+1} failed ({response.status_code}). '
                  f'Waiting {wait}s before retry...')
            time.sleep(wait)
        else:
            response.raise_for_status()  # non-retryable error

    raise Exception(f'Failed after {max_retries} retries')

การเพิ่มการหน่วงเวลาแบบสุ่ม

import requests
import time
import random

def get_with_jittered_backoff(url, headers, max_retries=5):
    for attempt in range(max_retries):
        response = requests.get(url, headers=headers, timeout=(5, 30))

        if response.status_code == 200:
            return response.json()

        if response.status_code in (429, 500, 502, 503):
            base_wait = 2 ** attempt
            # Add random jitter: actual wait is 50%-100% of base
            jitter = random.uniform(0.5, 1.0)
            wait = base_wait * jitter
            print(f'Waiting {wait:.1f}s (attempt {attempt+1})')
            time.sleep(wait)
        else:
            response.raise_for_status()

    raise Exception(f'Failed after {max_retries} retries')

ไลบรารี tenacity

from tenacity import (
    retry, stop_after_attempt, wait_exponential,
    retry_if_exception_type, before_sleep_log
)
import requests
import logging

logger = logging.getLogger(__name__)

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    retry=retry_if_exception_type(requests.exceptions.HTTPError),
    before_sleep=before_sleep_log(logger, logging.WARNING)
)
def fetch_data(url, headers):
    response = requests.get(url, headers=headers, timeout=(5, 30))
    if response.status_code == 429:
        response.raise_for_status()  # triggers retry
    response.raise_for_status()
    return response.json()

tenacity กับเงื่อนไขการลองใหม่แบบกำหนดเอง

from tenacity import (
    retry, stop_after_attempt, wait_exponential,
    retry_if_result
)
import requests

def is_retryable_response(response):
    return response.status_code in (429, 500, 502, 503, 504)

@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=2, min=2, max=30),
    retry=retry_if_result(is_retryable_response)
)
def resilient_get(url, headers):
    response = requests.get(url, headers=headers, timeout=(5, 30))
    return response  # retry logic inspects the response object

# Usage
response = resilient_get(
    'https://api.example.com/data',
    {'Authorization': 'Bearer YOUR_KEY'}
)
data = response.json()

การจัดการการจำกัดอัตราเชิงรุก

import requests
import time

class RateLimitAwareClient:
    def __init__(self, base_url, api_key):
        self.base_url = base_url
        self.headers = {'Authorization': f'Bearer {api_key}'}
        self.remaining = 1000  # assume generous limit

    def get(self, path):
        # Proactively slow down if nearly exhausted
        if self.remaining < 10:
            print('Rate limit nearly exhausted, sleeping 5s...')
            time.sleep(5)

        response = requests.get(
            f'{self.base_url}{path}', headers=self.headers
        )

        # Update remaining from response headers
        remaining_str = response.headers.get('X-RateLimit-Remaining')
        if remaining_str:
            self.remaining = int(remaining_str)

        response.raise_for_status()
        return response.json()

จำนวนครั้งลองใหม่สูงสุดและการยอมแพ้

import requests
import time

class MaxRetriesExceeded(Exception):
    def __init__(self, url, attempts, last_status):
        self.url = url
        self.attempts = attempts
        self.last_status = last_status
        super().__init__(
            f'Failed {url} after {attempts} attempts '
            f'(last status: {last_status})'
        )

def fetch_with_limit(url, headers, max_retries=3):
    last_response = None
    for attempt in range(max_retries):
        last_response = requests.get(url, headers=headers)
        if last_response.status_code == 200:
            return last_response.json()
        time.sleep(2 ** attempt)
    raise MaxRetriesExceeded(url, max_retries, last_response.status_code)

รูปแบบตัวตัดวงจร

import time

class CircuitBreaker:
    CLOSED, OPEN, HALF_OPEN = 'closed', 'open', 'half_open'

    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.state = self.CLOSED
        self.failures = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.opened_at = None

    def call(self, func, *args, **kwargs):
        if self.state == self.OPEN:
            if time.time() - self.opened_at > self.recovery_timeout:
                self.state = self.HALF_OPEN
            else:
                raise Exception('Circuit OPEN — service unavailable')
        try:
            result = func(*args, **kwargs)
            self.failures = 0
            self.state = self.CLOSED
            return result
        except Exception as e:
            self.failures += 1
            if self.failures >= self.failure_threshold:
                self.state = self.OPEN
                self.opened_at = time.time()
                print(f'Circuit OPENED after {self.failures} failures')
            raise

# --- demo ---
def flaky():
    raise ValueError('upstream 500')

def works():
    return 'ok'

cb = CircuitBreaker(failure_threshold=3, recovery_timeout=60)
for i in range(3):
    try:
        cb.call(flaky)
    except Exception as e:
        print(f'call {i+1} failed: {e}')
print(f'Breaker state after 3 failures: {cb.state}')
try:
    cb.call(flaky)
except Exception as e:
    print(f'Rejected without calling flaky(): {e}')

การจัดคิวคำขอให้อยู่ภายในขีดจำกัด

import requests
import time

def batch_requests(urls, headers, calls_per_minute=60):
    interval = 60.0 / calls_per_minute  # seconds between calls
    results = []

    for i, url in enumerate(urls):
        start = time.time()

        response = requests.get(url, headers=headers, timeout=(5, 30))
        response.raise_for_status()
        results.append(response.json())

        print(f'Processed {i+1}/{len(urls)}')

        # Sleep for remaining time in the interval
        elapsed = time.time() - start
        sleep_time = interval - elapsed
        if sleep_time > 0:
            time.sleep(sleep_time)

    return results

การผสานตรรกะการลองใหม่กับส่วนหัวการหน่วงเวลา

import requests
import time
import random

def smart_retry(url, headers, max_retries=5):
    for attempt in range(max_retries):
        response = requests.get(url, headers=headers, timeout=(5, 30))

        if response.status_code == 200:
            return response.json()

        if response.status_code == 429:
            # Use Retry-After if provided, else exponential backoff
            retry_after = response.headers.get('Retry-After')
            if retry_after:
                wait = int(retry_after)
            else:
                wait = (2 ** attempt) + random.uniform(0, 1)
            print(f'429 rate limit. Waiting {wait:.1f}s...')
            time.sleep(wait)

        elif response.status_code >= 500:
            wait = (2 ** attempt) + random.uniform(0, 1)
            print(f'Server error {response.status_code}. Waiting {wait:.1f}s...')
            time.sleep(wait)

        else:
            response.raise_for_status()  # non-retryable

    raise Exception(f'Gave up after {max_retries} attempts')

ตรวจสอบความเข้าใจอย่างรวดเร็ว: การหน่วงเวลาแบบทวีคูณ

สรุปการจำกัดอัตราและการลองใหม่