معالجة أخطاء API

#!/bin/sh
# POSIX shell, curl 8+, jq 1.7. Credentials stay in environment variables.
set -eu
: "${SHOPENTLY_SHOP_URL:?}"
base=${SHOPENTLY_SHOP_URL%/}
case "$base" in https://*) ;; *) echo 'Use an HTTPS origin' >&2; exit 1;; esac
authority=${base#https://}
case "$authority" in ''|*/*|*\?*|*\#*|*@*) echo 'Use an HTTPS origin' >&2; exit 1;; esac
work=$(mktemp -d)
trap 'rm -rf "$work"' EXIT HUP INT TERM
chmod 700 "$work"
encode() { printf '%s' "$1" | jq -sRr @uri; }
api() {
    method=$1; path=$2; body=${3-}; key=${4-}
    scheme=Bearer; credential=${SHOPENTLY_ACCESS_TOKEN-}
    if [ "$path" = /api/v1/auth/token ]; then scheme=ApiKey; credential=${SHOPENTLY_API_KEY-}; fi
    [ -n "$credential" ] || { echo 'Missing credential' >&2; return 1; }
    set -- --disable --globoff --silent --show-error --proto '=https' --connect-timeout 5 --max-time 30 \
        --request "$method" --header "Authorization: $scheme $credential" \
        --dump-header "$work/headers" --output "$work/body" --write-out '%{http_code}'
    if [ -n "$key" ]; then set -- "$@" --header "Idempotency-Key: $key"; fi
    if [ -n "$body" ]; then set -- "$@" --header 'Content-Type: application/json' --data-raw "$body"; fi
    status=$(curl "$@" --url "$base$path") || return 1
    case "$status" in
        2??) if [ -s "$work/body" ]; then jq -e . "$work/body"; else printf 'null\n'; fi ;;
        *) printf 'HTTP %s\n' "$status" >&2
           jq -c '.error | {code, fields}' "$work/body" >&2 2>/dev/null || true
           return 1 ;;
    esac
}

# Only this read is retried. Writes need a persisted key and unchanged body.
attempt=0
while :; do
    if api GET /api/v1/context; then break; fi
    attempt=$((attempt + 1))
    [ "$attempt" -lt 3 ] || exit 1
    case "${status-}" in 429|502|503|504) ;; *) exit 1;; esac
    retry=$(awk 'tolower($1) == "retry-after:" {gsub("\r","",$2); print $2}' "$work/headers")
    case "$retry" in '' ) retry=$attempt;; *[!0-9]* ) exit 1;; esac
    [ "$retry" -le 30 ] || exit 1
    sleep "$((retry + 1))"
done

<?php
declare(strict_types=1);
// PHP 8.3+, 64-bit, ext-curl. Credentials stay in environment variables.
function env(string $name): string {
    $value = getenv($name);
    if ($value === false || $value === '' || strpbrk($value, "\r\n") !== false) {
        throw new RuntimeException('Invalid environment variable: ' . $name);
    }
    return $value;
}
$base = rtrim(env('SHOPENTLY_SHOP_URL'), '/');
$origin = parse_url($base);
if (!$origin || ($origin['scheme'] ?? '') !== 'https' || empty($origin['host']) ||
    isset($origin['user']) || isset($origin['pass']) || isset($origin['path']) ||
    isset($origin['query']) || isset($origin['fragment']) || PHP_INT_SIZE !== 8) {
    throw new RuntimeException('Use a 64-bit PHP runtime and an HTTPS origin');
}
class ApiError extends RuntimeException {
    public function __construct(public int $status, public array $headers, public array $error) {
        parent::__construct('HTTP ' . $status . '; code=' . ($error['code'] ?? '') .
            '; request_id=' . ($headers['x-request-id'] ?? ''));
    }
}
function api(string $method, string $path, array $query = [], ?array $body = null, ?string $key = null, bool $apiKey = false): ?array {
    global $base;
    $credential = env($apiKey ? 'SHOPENTLY_API_KEY' : 'SHOPENTLY_ACCESS_TOKEN');
    $headers = ['Authorization: ' . ($apiKey ? 'ApiKey ' : 'Bearer ') . $credential];
    if ($key !== null) $headers[] = 'Idempotency-Key: ' . $key;
    if ($body !== null) $headers[] = 'Content-Type: application/json';
    $url = $base . $path . ($query ? '?' . http_build_query($query, '', '&', PHP_QUERY_RFC3986) : '');
    $responseHeaders = [];
    $h = curl_init($url);
    curl_setopt_array($h, [
        CURLOPT_CUSTOMREQUEST => $method, CURLOPT_HTTPHEADER => $headers,
        CURLOPT_RETURNTRANSFER => true, CURLOPT_FOLLOWLOCATION => false,
        CURLOPT_PROTOCOLS => CURLPROTO_HTTPS, CURLOPT_SSL_VERIFYPEER => true,
        CURLOPT_SSL_VERIFYHOST => 2, CURLOPT_CONNECTTIMEOUT => 5, CURLOPT_TIMEOUT => 30,
        CURLOPT_HEADERFUNCTION => static function ($h, string $line) use (&$responseHeaders): int {
            if (str_contains($line, ':')) {
                [$name, $value] = explode(':', $line, 2);
                $responseHeaders[strtolower(trim($name))] = trim($value);
            }
            return strlen($line);
        },
    ]);
    if ($body !== null) {
        $json = $apiKey && $body === [] ? '{}' : json_encode($body, JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE);
        curl_setopt($h, CURLOPT_POSTFIELDS, $json);
    }
    $raw = curl_exec($h);
    $status = curl_getinfo($h, CURLINFO_RESPONSE_CODE);
    $error = curl_error($h);
    curl_close($h);
    if ($raw === false) throw new RuntimeException('Transport error: ' . $error);
    if ($status < 200 || $status >= 300) {
        $data = json_decode($raw, true);
        throw new ApiError($status, $responseHeaders, is_array($data) ? ($data['error'] ?? []) : []);
    }
    return $raw === '' ? null : json_decode($raw, true, 512, JSON_THROW_ON_ERROR | JSON_BIGINT_AS_STRING);
}
function output(mixed $value): void {
    echo json_encode($value, JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE), PHP_EOL;
}
// Only this read is retried. Writes require a persisted key and unchanged body.
for ($attempt = 0; $attempt < 3; $attempt++) {
    try { output(api('GET', '/api/v1/context')); break; }
    catch (ApiError $error) {
        if (!in_array($error->status, [429, 502, 503, 504], true) || $attempt === 2) throw $error;
        $retry = $error->headers['retry-after'] ?? '';
        if ($retry !== '' && !preg_match('/^[0-9]+$/', $retry)) throw $error;
        $delay = $retry === '' ? 2 ** $attempt : (int)$retry;
        if ($delay > 30) throw $error;
        usleep(($delay * 1000000) + random_int(0, 1000000));
    }
}

"""Run with Python 3.12 and requests. Credentials stay in environment variables."""
import json
import os
import random
import time
from urllib.parse import quote, urlencode, urlsplit
import requests

def env(name):
    value = os.environ[name]
    if not value or "\r" in value or "\n" in value:
        raise ValueError("Invalid environment variable: " + name)
    return value

base = env("SHOPENTLY_SHOP_URL").rstrip("/")
origin = urlsplit(base)
if origin.scheme != "https" or not origin.hostname or origin.username or origin.password or origin.path or origin.query or origin.fragment:
    raise ValueError("SHOPENTLY_SHOP_URL must be an HTTPS origin")

class ApiError(RuntimeError):
    def __init__(self, response):
        self.status = response.status_code
        self.request_id = response.headers.get("X-Request-ID", "")
        self.retry_after = response.headers.get("Retry-After", "")
        try:
            self.error = response.json().get("error", {})
        except (ValueError, AttributeError):
            self.error = {}
        super().__init__(f"HTTP {self.status}; code={self.error.get('code', '')}; request_id={self.request_id}")

def api(method, path, *, query=None, body=None, key=None, api_key=False):
    credential = env("SHOPENTLY_API_KEY" if api_key else "SHOPENTLY_ACCESS_TOKEN")
    headers = {"Authorization": ("ApiKey " if api_key else "Bearer ") + credential}
    if key:
        headers["Idempotency-Key"] = key
    if body is not None:
        headers["Content-Type"] = "application/json"
    response = requests.request(method, base + path, params=query, headers=headers,
        data=json.dumps(body, ensure_ascii=False, allow_nan=False).encode() if body is not None else None,
        timeout=(5, 30), allow_redirects=False)
    if not 200 <= response.status_code < 300:
        raise ApiError(response)
    return response.json() if response.content else None

def segment(value):
    return quote(value, safe="")

# Only this read is retried. Writes require a persisted key and unchanged body.
for attempt in range(3):
    try:
        print(json.dumps(api("GET", "/api/v1/context"), ensure_ascii=False))
        break
    except ApiError as error:
        if error.status not in (429, 502, 503, 504) or attempt == 2:
            raise
        if error.retry_after and not error.retry_after.isdigit():
            raise
        delay = int(error.retry_after or 2 ** attempt)
        if delay > 30:
            raise  # Schedule a later job; do not retry earlier than Retry-After.
        time.sleep(delay + random.uniform(0, 1))