速率限制

{
  "type": "https://docs.livingsecurity.com/developers/v1/errors/rate_limited",
  "title": "Too Many Requests",
  "status": 429,
  "detail": "Rate limit exceeded. Retry after the Retry-After interval.",
  "code": "rate_limited",
  "request_id": "req_...",
  "doc_url": "https://docs.livingsecurity.com/developers/v1/errors/rate_limited",
  "retryable": true
}

import os
import random
import time
import requests

# US: BASE = "https://api.us-east-1.livingsecurity.ai/v1"
BASE = "https://api.eu-west-1.livingsecurity.ai/v1"
AUTH = ("apikey", os.environ["LIVING_SECURITY_API_KEY"])

def get_with_retry(path, params=None, max_attempts=5):
    if not 1 <= max_attempts <= 5:
        raise ValueError("max_attempts must be between 1 and 5")
    for attempt in range(max_attempts):
        response = requests.get(
            f"{BASE}{path}", auth=AUTH, params=params,
            headers={"Living-Security-Version": "2026-07-01"},
            timeout=(5, 30), allow_redirects=False,
        )
        if response.status_code != 429:
            response.raise_for_status()
            if not 200 <= response.status_code < 300:
                raise RuntimeError("Unexpected redirect; check your regional API URL")
            return response

        # The API returns Retry-After as a whole number of seconds.
        raw_wait = response.headers.get("Retry-After", "")
        request_id = response.headers.get("Request-Id", "unknown")
        response.close()
        if attempt == max_attempts - 1:
            raise RuntimeError(f"Retry budget exhausted; request_id={request_id}")
        if not raw_wait.isascii() or not raw_wait.isdigit() or len(raw_wait) > 6:
            raise RuntimeError(f"Invalid Retry-After; request_id={request_id}")

        wait = max(int(raw_wait), min(2 ** attempt, 30))
        delay = wait + random.uniform(0, 1)
        if delay > 60:
            raise RuntimeError(f"Retry-After exceeds wait budget; request_id={request_id}")
        time.sleep(delay)