更新 · 库存 — 库存

: "${SHOPENTLY_SHOP_URL:?}"

curl --disable --globoff --fail-with-body --silent --show-error \
  --proto '=https' --connect-timeout 5 --max-time 30 \
  --request PATCH \
  --header "Authorization: Bearer ${SHOPENTLY_ACCESS_TOKEN:?}" \
  --header 'Idempotency-Key: example-update_stock_api_v1_inventory_stock__stock_id__patch' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "expected_version": 1,
  "locations": [
    {
      "location_id": "example",
      "on_hand": 1
    }
  ],
  "reason": "Example"
}' \
  --url "${SHOPENTLY_SHOP_URL%/}"/api/v1/inventory/stock/stock_v1_WyJleGFtcGxlIiwicHJvZHVjdCIsImV4YW1wbGUiXQ

<?php
declare(strict_types=1);

$baseUrl = rtrim(getenv('SHOPENTLY_SHOP_URL') ?: '', '/');
$origin = parse_url($baseUrl);
if (!$origin || ($origin['scheme'] ?? '') !== 'https' || empty($origin['host']) ||
    isset($origin['user'], $origin['pass']) || isset($origin['user']) || isset($origin['path']) ||
    isset($origin['query']) || isset($origin['fragment'])) {
    throw new RuntimeException('SHOPENTLY_SHOP_URL must be an HTTPS origin');
}
$credential = getenv('SHOPENTLY_ACCESS_TOKEN') ?: '';
if ($credential === '' || strpbrk($credential, "\r\n") !== false) {
    throw new RuntimeException('Invalid credential');
}
$headers = ['Idempotency-Key: example-update_stock_api_v1_inventory_stock__stock_id__patch', 'Content-Type: application/json'];
$headers[] = 'Authorization: Bearer ' . $credential;
$responseHeaders = [];
$handle = curl_init($baseUrl . '/api/v1/inventory/stock/stock_v1_WyJleGFtcGxlIiwicHJvZHVjdCIsImV4YW1wbGUiXQ');
curl_setopt_array($handle, [
    CURLOPT_CUSTOMREQUEST => 'PATCH',
    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 ($handle, string $line) use (&$responseHeaders): int {
        if (str_contains($line, ':')) {
            [$name, $value] = explode(':', $line, 2);
            $responseHeaders[strtolower(trim($name))] = trim($value);
        }
        return strlen($line);
    },
]);
curl_setopt($handle, CURLOPT_POSTFIELDS, '{
  "expected_version": 1,
  "locations": [
    {
      "location_id": "example",
      "on_hand": 1
    }
  ],
  "reason": "Example"
}');
$raw = curl_exec($handle);
$status = curl_getinfo($handle, CURLINFO_RESPONSE_CODE);
$error = curl_error($handle);
curl_close($handle);
if ($raw === false) {
    throw new RuntimeException('Transport error: ' . $error);
}
if ($status < 200 || $status >= 300) {
    throw new RuntimeException('HTTP ' . $status . '; request_id=' .
        ($responseHeaders['x-request-id'] ?? '') . '; retry_after=' . ($responseHeaders['retry-after'] ?? ''));
}
$result = $raw === '' ? null : json_decode($raw, true, 512, JSON_THROW_ON_ERROR | JSON_BIGINT_AS_STRING);
echo json_encode($result, JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT | JSON_UNESCAPED_UNICODE), PHP_EOL;

from contextlib import ExitStack
import json
import os
from urllib.parse import urlsplit

import requests  # python -m pip install requests

base_url = os.environ["SHOPENTLY_SHOP_URL"].rstrip("/")
origin = urlsplit(base_url)
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")
credential = os.environ['SHOPENTLY_ACCESS_TOKEN']
if not credential or "\r" in credential or "\n" in credential:
    raise ValueError("Invalid credential")
headers = {'Idempotency-Key': 'example-update_stock_api_v1_inventory_stock__stock_id__patch'}
headers["Authorization"] = 'Bearer' + " " + credential
body = '{\n  "expected_version": 1,\n  "locations": [\n    {\n      "location_id": "example",\n      "on_hand": 1\n    }\n  ],\n  "reason": "Example"\n}'
if body is not None:
    headers["Content-Type"] = "application/json"

with ExitStack() as stack:
    session = stack.enter_context(requests.Session())
    files = None
    if False:
        image = stack.enter_context(open(os.environ["SHOPENTLY_IMAGE_FILE"], "rb"))
        if os.fstat(image.fileno()).st_size > 5242880:
            raise ValueError("Image exceeds 5 MiB")
        files = {"file": ("article.png", image, "image/png")}
    response = session.request(
        'PATCH', base_url + '/api/v1/inventory/stock/stock_v1_WyJleGFtcGxlIiwicHJvZHVjdCIsImV4YW1wbGUiXQ', headers=headers,
        data=body.encode("utf-8") if body is not None else None, files=files,
        timeout=(5, 30), allow_redirects=False,
    )
    if not 200 <= response.status_code < 300:
        raise RuntimeError(f"HTTP {response.status_code}; request_id={response.headers.get('X-Request-ID', '')}; retry_after={response.headers.get('Retry-After', '')}")
    if False:
        if response.headers.get("Content-Type") != "image/webp" or len(response.content) > 5242880:
            raise ValueError("Unexpected image response")
        with open(os.environ["SHOPENTLY_OUTPUT_FILE"], "xb") as output:
            output.write(response.content)
    else:
        result = response.json() if response.content else None
        print(json.dumps(result, ensure_ascii=False, indent=2))