Przykłady kodu | AMPLIFIER Baza Wiedzy

Przykłady pokazują cały przebieg integracji: logowanie, pobranie listy stronami z ponowieniem po odpowiedzi 429 i wycenę koszyka w trybie playground. W przykładach zastąp {amper_ws}, {api_key}, login, hasło i ID klienta swoimi danymi.

Zasady, na których opierają się przykłady: Uwierzytelnianie w API zewnętrznym, Stronicowanie, limity i kody odpowiedzi, Tworzenie zamówienia „create-order”.

curl

Odczyt z kluczem API:

curl "https://{amper_ws}/external/v1/products/?api_key={api_key}&limit=500&offset=0&ordering=id"

Logowanie integratora i odczyt z tokenem:

curl -X POST https://{amper_ws}/external/v1/auth/ \
  -H "Content-Type: application/json" \
  -d '{"username": "login", "password": "haslo"}'

curl "https://{amper_ws}/external/v1/promotions/?limit=50&offset=0" \
  -H "Authorization: Bearer <access_token>"

Wycena koszyka:

curl -X POST https://{amper_ws}/external/v1/create-order \
  -H "Authorization: Bearer <access_token>" \
  -H "Content-Type: application/json" \
  -d '{"customer_id": 42, "playground": true, "lines": [{"product_id": 101, "quantity": 12, "suggested_price": null}]}'

Python

Klient z biblioteką requests: loguje się, odświeża token przed wygaśnięciem i po odpowiedzi 401, czeka po odpowiedzi 429, ponawia odczyt po błędzie serwera (zamówień nie ponawia) i pobiera listy stronami. Działa też z samym kluczem API.

import time

import requests


class AmperExternalApi:
    """API zewnętrzne AMPER: logowanie, odświeżanie tokena, stronicowanie i ponawianie po 429."""

    def __init__(self, host, username=None, password=None, api_key=None, timeout=120):
        self.base_url = f"https://{host}/external/v1"
        self.username = username
        self.password = password
        self.api_key = api_key
        self.timeout = timeout
        self.session = requests.Session()
        self.access_token = None
        self.refresh_token = None
        self.expires_at = 0

    def _save_tokens(self, data):
        self.access_token = data["access_token"]
        self.refresh_token = data["refresh_token"]
        self.expires_at = time.time() + data["expires_in"] - 30  # 30 s zapasu

    def login(self):
        response = self.session.post(f"{self.base_url}/auth/", timeout=self.timeout,
                                     json={"username": self.username, "password": self.password})
        response.raise_for_status()
        self._save_tokens(response.json())

    def refresh(self):
        response = self.session.post(f"{self.base_url}/auth/refresh/", timeout=self.timeout,
                                     json={"refresh_token": self.refresh_token})
        if response.status_code == 400:  # refresh_token wygasł – logujemy się ponownie
            self.login()
            return
        response.raise_for_status()
        self._save_tokens(response.json())

    def _credentials(self):
        """(nagłówki, parametry adresu) z danymi dostępu."""
        if self.api_key:
            return {}, {"api_key": self.api_key}
        if not self.access_token:
            self.login()
        elif time.time() >= self.expires_at:
            self.refresh()
        return {"Authorization": f"Bearer {self.access_token}"}, {}

    def request(self, method, path, params=None, json=None):
        refreshed = False
        for attempt in range(5):
            headers, auth_params = self._credentials()
            response = self.session.request(method, f"{self.base_url}/{path}", headers=headers,
                                            params={**(params or {}), **auth_params}, json=json,
                                            timeout=self.timeout)
            if response.status_code == 429:  # limit zapytań – czekamy i ponawiamy
                time.sleep(int(response.headers.get("Retry-After", 1)))
                continue
            if response.status_code == 401 and not self.api_key and not refreshed:
                self.expires_at = 0  # token wygasł – odświeżamy i ponawiamy raz
                refreshed = True
                continue
            if response.status_code >= 500 and method == "GET":  # zamówień nie ponawiamy
                time.sleep(2 ** attempt)
                continue
            break
        response.raise_for_status()
        return response.json()

    def get_all(self, path, page_size=500, **params):
        """Cała lista, strona po stronie."""
        items, offset = [], 0
        while True:
            page = self.request("GET", path, params={**params, "limit": page_size, "offset": offset})
            items.extend(page["results"])
            if not page["next"]:
                return items
            offset += page_size

Użycie:

api = AmperExternalApi("{amper_ws}", username="login", password="haslo")
# tylko odczyt: AmperExternalApi("{amper_ws}", api_key="{api_key}")

categories = api.get_all("categories/")
products = api.get_all("products/", ordering="id")
stocks = api.get_all("stocks/")
promotions = api.get_all("promotions/", page_size=50)

lines = [{"product_id": 101, "quantity": 12, "suggested_price": None}]

quote = api.request("POST", "create-order", json={"customer_id": 42, "playground": True, "lines": lines})
for line in quote["lines"]:
    print(line["product_id"], line["quantity"], line["given_price"] or line["customer_price"])

order = api.request("POST", "create-order", json={"customer_id": 42, "lines": lines})
if not order["order_id"]:
    raise RuntimeError("Zamówienie nie zostało utworzone")

C#

.NET 6 i nowsze, bez dodatkowych bibliotek. Treść żądań to StringContent – wysyła nagłówek Content-Length.

using System.Net;
using System.Net.Http.Headers;
using System.Net.Http.Json;
using System.Text;
using System.Text.Json;

var baseUrl = "https://{amper_ws}/external/v1";
using var client = new HttpClient { Timeout = TimeSpan.FromMinutes(2) };

// Treść JSON (z nagłówkiem Content-Length)
StringContent Json(object body) =>
    new(JsonSerializer.Serialize(body), Encoding.UTF8, "application/json");

// Zapytanie z ponowieniem po 429 (limit 3 jednoczesnych zapytań)
async Task<JsonElement> Send(Func<HttpRequestMessage> request)
{
    for (var attempt = 0; attempt < 5; attempt++)
    {
        using var response = await client.SendAsync(request());
        if (response.StatusCode == HttpStatusCode.TooManyRequests)
        {
            await Task.Delay(response.Headers.RetryAfter?.Delta ?? TimeSpan.FromSeconds(1));
            continue;
        }
        response.EnsureSuccessStatusCode();
        return await response.Content.ReadFromJsonAsync<JsonElement>();
    }
    throw new HttpRequestException("Przekroczony limit zapytań");
}

// Logowanie integratora
var token = await Send(() => new HttpRequestMessage(HttpMethod.Post, $"{baseUrl}/auth/")
{
    Content = Json(new { username = "login", password = "haslo" })
});
client.DefaultRequestHeaders.Authorization =
    new AuthenticationHeaderValue("Bearer", token.GetProperty("access_token").GetString());

// Cała lista produktów, strona po stronie
var products = new List<JsonElement>();
for (var offset = 0; ; offset += 500)
{
    var page = await Send(() => new HttpRequestMessage(HttpMethod.Get,
        $"{baseUrl}/products/?limit=500&offset={offset}&ordering=id"));
    products.AddRange(page.GetProperty("results").EnumerateArray());
    if (page.GetProperty("next").ValueKind == JsonValueKind.Null) break;
}
Console.WriteLine($"Produkty: {products.Count}");

// Wycena koszyka (playground)
var quote = await Send(() => new HttpRequestMessage(HttpMethod.Post, $"{baseUrl}/create-order")
{
    Content = Json(new
    {
        customer_id = 42,
        playground = true,
        lines = new[] { new { product_id = 101, quantity = 12m, suggested_price = (decimal?)null } }
    })
});
foreach (var line in quote.GetProperty("lines").EnumerateArray())
    Console.WriteLine($"{line.GetProperty("product_id")} {line.GetProperty("customer_price")}");

Translatory ERP mogą zamiast tego korzystać z biblioteki AmperApiClient.NET.

Node.js

Node.js 18 i nowsze (wbudowany fetch), moduł ES (.mjs).

const BASE_URL = 'https://{amper_ws}/external/v1';
const sleep = (seconds) => new Promise((resolve) => setTimeout(resolve, seconds * 1000));

// Logowanie integratora: zwraca access_token
async function login(username, password) {
  const response = await fetch(`${BASE_URL}/auth/`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ username, password }),
  });
  if (!response.ok) throw new Error(`Logowanie: HTTP ${response.status}`);
  return (await response.json()).access_token;
}

// Zapytanie z ponowieniem po 429
async function call(path, token, options = {}) {
  for (let attempt = 0; attempt < 5; attempt++) {
    const response = await fetch(`${BASE_URL}/${path}`, {
      ...options,
      headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${token}`, ...options.headers },
    });
    if (response.status === 429) {
      await sleep(Number(response.headers.get('Retry-After') ?? 1));
      continue;
    }
    if (!response.ok) throw new Error(`${path}: HTTP ${response.status}`);
    return response.json();
  }
  throw new Error(`${path}: przekroczony limit zapytań`);
}

// Cała lista, strona po stronie
async function getAll(path, token, pageSize = 500, query = '') {
  const items = [];
  for (let offset = 0; ; offset += pageSize) {
    const page = await call(`${path}?limit=${pageSize}&offset=${offset}${query}`, token);
    items.push(...page.results);
    if (!page.next) return items;
  }
}

const token = await login('login', 'haslo');
const products = await getAll('products/', token, 500, '&ordering=id');
console.log(`Produkty: ${products.length}`);

const quote = await call('create-order', token, {
  method: 'POST',
  body: JSON.stringify({
    customer_id: 42,
    playground: true,
    lines: [{ product_id: 101, quantity: 12, suggested_price: null }],
  }),
});
for (const line of quote.lines) {
  console.log(line.product_id, line.quantity, line.given_price || line.customer_price);
}

PHP

PHP 8 z rozszerzeniem cURL.

<?php
const BASE_URL = 'https://{amper_ws}/external/v1';

function amper(string $method, string $path, ?string $token = null, ?array $body = null): array
{
    for ($attempt = 0; $attempt < 5; $attempt++) {
        $curl = curl_init(BASE_URL . '/' . $path);
        $headers = ['Content-Type: application/json'];
        if ($token !== null) {
            $headers[] = 'Authorization: Bearer ' . $token;
        }
        curl_setopt_array($curl, [
            CURLOPT_CUSTOMREQUEST => $method,
            CURLOPT_HTTPHEADER => $headers,
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_TIMEOUT => 120,
        ]);
        if ($body !== null) {
            curl_setopt($curl, CURLOPT_POSTFIELDS, json_encode($body));
        }
        $response = curl_exec($curl);
        $status = curl_getinfo($curl, CURLINFO_RESPONSE_CODE);
        if ($status === 429) { // limit zapytań – czekamy i ponawiamy
            sleep(1);
            continue;
        }
        if ($response === false || $status >= 400) {
            throw new RuntimeException("$method $path: HTTP $status");
        }
        return json_decode($response, true);
    }
    throw new RuntimeException("$method $path: przekroczony limit zapytań");
}

// Logowanie integratora
$token = amper('POST', 'auth/', null, ['username' => 'login', 'password' => 'haslo'])['access_token'];

// Cała lista produktów, strona po stronie
$products = [];
for ($offset = 0; ; $offset += 500) {
    $page = amper('GET', "products/?limit=500&offset=$offset&ordering=id", $token);
    $products = array_merge($products, $page['results']);
    if ($page['next'] === null) {
        break;
    }
}
echo 'Produkty: ', count($products), PHP_EOL;

// Wycena koszyka (playground)
$quote = amper('POST', 'create-order', $token, [
    'customer_id' => 42,
    'playground' => true,
    'lines' => [['product_id' => 101, 'quantity' => 12, 'suggested_price' => null]],
]);
foreach ($quote['lines'] as $line) {
    echo $line['product_id'], ' ', $line['given_price'] ?: $line['customer_price'], PHP_EOL;
}