API Error Codes

{
  "detail": {
    "error": {
      "code": "ERROR_CODE",
      "message": "Human-readable description of what went wrong"
    }
  }
}

{
  "detail": {
    "error": {
      "code": "UNAUTHORIZED",
      "message": "Missing API key"
    }
  }
}

{
  "error": "Rate limit exceeded: 60 per 1 minute"
}

{
  "detail": [
    {
      "type": "missing",
      "loc": ["body", "service"],
      "msg": "Field required",
      "input": {}
    }
  ]
}

{
  "detail": {
    "error": {
      "code": "MAINTENANCE",
      "message": "New purchases are paused for a short maintenance. Numbers you already rented still receive their codes. Your balance is safe. Please try again in a few minutes."
    }
  }
}

import random
import time
from datetime import datetime

import requests

API = "https://api.diddysms.com/v1"
HEADERS = {"Authorization": "Bearer dsk_your_api_key_here"}
TEMPORARY = {429, 502, 503, 504, 520, 521, 522, 523, 524}


def is_temporary(resp):
    if resp is None:  # network error or timeout
        return True
    if resp.status_code == 429 and "MAX_RENTALS" in resp.text:
        return False  # not a rate limit: an order has to finish first
    return resp.status_code in TEMPORARY


def wait(resp, attempt):
    """Retry-After when the API sends it, else 2, 4, 8 ... 60 s, plus jitter."""
    header = resp.headers.get("Retry-After") if resp is not None else None
    delay = int(header) if header and header.isdigit() else min(60, 2 ** (attempt + 1))
    time.sleep(delay + random.random())


def request(method, path, **kwargs):
    """Reads, cancel and complete: safe to repeat."""
    for attempt in range(12):
        try:
            resp = requests.request(
                method, f"{API}{path}", headers=HEADERS, timeout=30, **kwargs
            )
        except requests.RequestException:
            resp = None
        if not is_temporary(resp):
            return resp
        wait(resp, attempt)
    raise RuntimeError(f"{method} {path}: still failing, try again later")


def buy(service):
    """A buy is not safe to repeat blindly: look for the order first."""
    started = time.time()
    for attempt in range(12):
        try:
            resp = requests.post(
                f"{API}/orders", headers=HEADERS, json={"service": service}, timeout=60
            )
        except requests.RequestException:
            resp = None
        if resp is not None and resp.status_code == 201:
            return resp.json()["order"]
        if not is_temporary(resp):
            resp.raise_for_status()  # 4xx: fix the cause, do not repeat
        if resp is None or resp.status_code not in (429, 503):
            # Timeout, 502, 504 or 52x: the order may exist. (If you run buys
            # in parallel, also check another buy has not claimed it.)
            for order in request("GET", "/orders").json()["orders"]:
                created = datetime.fromisoformat(order["created_at"].replace("Z", "+00:00"))
                if order["service"] == service and created.timestamp() >= started - 5:
                    return order
        wait(resp, attempt)
    raise RuntimeError(f"buy {service}: still failing, try again later")