Authentication

const { SPARKLAYER_BASE_URL, SPARKLAYER_SITE_ID, SPARKLAYER_CLIENT_ID, SPARKLAYER_CLIENT_SECRET } =
  process.env;

let token = null; // { value, expiresAt }

async function getToken() {
  // reuse the token until a minute before it expires
  if (token && Date.now() < token.expiresAt - 60_000) return token.value;
  const response = await fetch(`${SPARKLAYER_BASE_URL}/api/auth/token`, {
    method: 'POST',
    headers: { 'Site-Id': SPARKLAYER_SITE_ID, 'Content-Type': 'application/json' },
    body: JSON.stringify({
      grant_type: 'client_credentials',
      client_id: SPARKLAYER_CLIENT_ID,
      client_secret: SPARKLAYER_CLIENT_SECRET,
    }),
  });
  if (!response.ok) throw new Error(`Token request failed: ${response.status}`);
  const body = await response.json();
  token = { value: body.access_token, expiresAt: Date.now() + body.expires_in * 1000 };
  return token.value;
}

/** Calls a SparkLayer API path, e.g. sparklayer('/api/v2/price-lists'). */
export async function sparklayer(path, init = {}, retried = false) {
  const response = await fetch(`${SPARKLAYER_BASE_URL}${path}`, {
    ...init,
    headers: {
      'Site-Id': SPARKLAYER_SITE_ID,
      Authorization: `Bearer ${await getToken()}`,
      'User-Agent': 'acme-erp-sync/1.2',
      ...(init.body ? { 'Content-Type': 'application/json' } : {}),
      ...init.headers,
    },
  });
  if (response.status === 401 && !retried) {
    token = null; // the token was rejected: get a new one and try once more
    return sparklayer(path, init, true);
  }
  return response;
}

import os
import time

import requests

BASE_URL = os.environ["SPARKLAYER_BASE_URL"]
SITE_ID = os.environ["SPARKLAYER_SITE_ID"]
_token = {"value": None, "expires_at": 0.0}

def get_token() -> str:
    # reuse the token until a minute before it expires
    if _token["value"] and time.time() < _token["expires_at"] - 60:
        return _token["value"]
    response = requests.post(
        f"{BASE_URL}/api/auth/token",
        headers={"Site-Id": SITE_ID},
        json={
            "grant_type": "client_credentials",
            "client_id": os.environ["SPARKLAYER_CLIENT_ID"],
            "client_secret": os.environ["SPARKLAYER_CLIENT_SECRET"],
        },
        timeout=30,
    )
    response.raise_for_status()
    body = response.json()
    _token.update(value=body["access_token"], expires_at=time.time() + body["expires_in"])
    return _token["value"]

def sparklayer(method: str, path: str, retried: bool = False, **kwargs) -> requests.Response:
    """Calls a SparkLayer API path, e.g. sparklayer("GET", "/api/v2/price-lists")."""
    headers = {
        "Site-Id": SITE_ID,
        "Authorization": f"Bearer {get_token()}",
        "User-Agent": "acme-erp-sync/1.2",
        **kwargs.pop("headers", {}),
    }
    response = requests.request(method, f"{BASE_URL}{path}", headers=headers, timeout=60, **kwargs)
    if response.status_code == 401 and not retried:
        _token["value"] = None  # the token was rejected: get a new one and try once more
        return sparklayer(method, path, retried=True, **kwargs)
    return response