Clients and SDKs

curl "https://api.brellium.com/v2/audits/a1b2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d" \
  --header "Authorization: Bearer $BRELLIUM_ACCESS_TOKEN"

const res = await fetch(
  "https://api.brellium.com/v2/audits/a1b2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d",
  { headers: { Authorization: `Bearer ${process.env.BRELLIUM_ACCESS_TOKEN}` } },
);
const body = await res.json();
if (!res.ok) throw new Error(`HTTP ${res.status}: ${JSON.stringify(body.error)}`);
console.log(body);

import BrelliumClient from "@brellium/api-client";

const client = new BrelliumClient({ oAuth2: process.env.BRELLIUM_ACCESS_TOKEN });

const audit = await client.audits.retrieve("a1b2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d");
console.log(audit);

import os

from brellium_api_client import BrelliumClient

client = BrelliumClient(o_auth2=os.environ["BRELLIUM_ACCESS_TOKEN"])

audit = client.audits.retrieve("a1b2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d")
print(audit.to_json())

// Keep credentials on the server; never in a browser or mobile app.
const BASE_URL = "https://api.brellium.com";
const CLIENT_ID = process.env.BRELLIUM_CLIENT_ID;
const CLIENT_SECRET = process.env.BRELLIUM_CLIENT_SECRET;
if (!CLIENT_ID || !CLIENT_SECRET) {
  throw new Error("Set BRELLIUM_CLIENT_ID and BRELLIUM_CLIENT_SECRET");
}

type TokenResponse = { access_token: string; expires_in: number };
type Page<T> = {
  object: "list";
  data: T[];
  has_more: boolean;
  total_count?: number;
};
type ApiError = {
  error: { type: string; code: string; message: string; request_id?: string };
};
type Audit = {
  audit_id: string;
  status: string;
  score: number | null;
  set_id: string;
  documents: string[];
};

let cached: { token: string; expiresAt: number } | undefined;

// 1. Mint once, cache until expires_in minus 300 s.
async function accessToken(): Promise<string> {
  if (cached && Date.now() < cached.expiresAt) return cached.token;
  const res = await fetch(`${BASE_URL}/v2/oauth/token`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      grant_type: "client_credentials",
      client_id: CLIENT_ID,
      client_secret: CLIENT_SECRET,
      audience: "https://api.brellium.dev",
    }),
  });
  if (!res.ok) throw new Error(`POST /v2/oauth/token failed: ${res.status}`);
  const body = (await res.json()) as TokenResponse;
  cached = {
    token: body.access_token,
    expiresAt: Date.now() + (body.expires_in - 300) * 1000,
  };
  return cached.token;
}

// 2. One request helper: bearer header, JSON body, wait Retry-After on 429, re-mint
//    once on 401.
async function request<T>(
  method: string,
  path: string,
  body?: unknown,
): Promise<T> {
  let reminted = false;
  for (let attempt = 0; ; attempt++) {
    const headers: Record<string, string> = {
      Authorization: `Bearer ${await accessToken()}`,
    };
    if (body !== undefined) headers["Content-Type"] = "application/json";
    const res = await fetch(`${BASE_URL}${path}`, {
      method,
      headers,
      body: body === undefined ? undefined : JSON.stringify(body),
    });
    if (res.status === 429 && attempt < 5) {
      const seconds = Number(res.headers.get("Retry-After") ?? "1");
      await new Promise((resolve) =>
        setTimeout(resolve, Math.max(seconds, 1) * 1000),
      );
      continue;
    }
    if (res.status === 401 && !reminted) {
      cached = undefined;
      reminted = true;
      continue;
    }
    if (res.status === 204) return undefined as unknown as T;
    const json = await res.json();
    if (!res.ok) {
      const { error } = json as ApiError;
      throw new Error(
        `${method} ${path}: ${res.status} ${error.code} ${error.message} ` +
          `(request_id ${error.request_id})`,
      );
    }
    return json as T;
  }
}

// 3. Page a list: limit=100, starting_after = the last item's id field, stop when
//    has_more is false.
//    idField names the resource's id: "audit_id", "user_id", "person_id" and so on.
async function* list<T>(
  path: string,
  idField: keyof T,
  query: Record<string, string> = {},
): AsyncGenerator<T> {
  let startingAfter: string | undefined;
  do {
    const params = new URLSearchParams({ ...query, limit: "100" });
    if (startingAfter) params.set("starting_after", startingAfter);
    const page = await request<Page<T>>("GET", `${path}?${params.toString()}`);
    yield* page.data;
    const last = page.data[page.data.length - 1];
    startingAfter = page.has_more && last ? String(last[idField]) : undefined;
  } while (startingAfter);
}

async function main(): Promise<void> {
  // Take the first complete audit from the list, then fetch it by id.
  for await (const item of list<Audit>("/v2/audits", "audit_id", {
    status: "complete",
  })) {
    const audit = await request<Audit>("GET", `/v2/audits/${item.audit_id}`);
    console.log(
      audit.audit_id,
      audit.status,
      audit.score,
      audit.set_id,
      audit.documents.length,
    );
    break;
  }
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});

# Keep credentials on the server; never in a browser or mobile app.
import os
import time
from collections.abc import Iterator
from typing import Any

import requests

BASE_URL = "https://api.brellium.com"


class Brellium:
    def __init__(self, client_id: str, client_secret: str) -> None:
        self._client_id = client_id
        self._client_secret = client_secret
        self._session = requests.Session()
        self._token: str | None = None
        self._expires_at = 0.0

    # 1. Mint once, cache until expires_in minus 300 s.
    def _access_token(self) -> str:
        if self._token and time.monotonic() < self._expires_at:
            return self._token
        r = self._session.post(
            f"{BASE_URL}/v2/oauth/token",
            json={
                "grant_type": "client_credentials",
                "client_id": self._client_id,
                "client_secret": self._client_secret,
                "audience": "https://api.brellium.dev",
            },
            timeout=30,
        )
        r.raise_for_status()
        body = r.json()
        self._token = body["access_token"]
        self._expires_at = time.monotonic() + body["expires_in"] - 300
        return self._token

    # 2. One request helper: bearer header, JSON body, wait Retry-After on 429,
    #    re-mint once on 401.
    def request(
        self,
        method: str,
        path: str,
        body: Any = None,
        params: dict[str, Any] | None = None,
    ) -> Any:
        reminted = False
        attempt = 0
        while True:
            r = self._session.request(
                method,
                f"{BASE_URL}{path}",
                params=params,
                json=body,
                headers={"Authorization": f"Bearer {self._access_token()}"},
                timeout=30,
            )
            if r.status_code == 429 and attempt < 5:
                attempt += 1
                time.sleep(max(float(r.headers.get("Retry-After", "1")), 1.0))
                continue
            if r.status_code == 401 and not reminted:
                self._token, reminted = None, True
                continue
            if r.status_code == 204:
                return None
            payload = r.json()
            if not r.ok:
                err = payload["error"]
                raise RuntimeError(
                    f"{method} {path}: {r.status_code} {err['code']} "
                    f"{err['message']} "
                    f"(request_id {err.get('request_id')})"
                )
            return payload

    # 3. Page a list: limit=100, starting_after = the last item's id field, stop
    #    when has_more is false.
    # id_field names the resource's id: "audit_id", "user_id", "person_id" and so
    # on.
    def iter_list(
        self, path: str, id_field: str, **filters: str
    ) -> Iterator[dict[str, Any]]:
        params: dict[str, Any] = {"limit": 100, **filters}
        while True:
            page = self.request("GET", path, params=params)
            yield from page["data"]
            if not page["has_more"] or not page["data"]:
                return
            params["starting_after"] = page["data"][-1][id_field]


if __name__ == "__main__":
    api = Brellium(
        os.environ["BRELLIUM_CLIENT_ID"], os.environ["BRELLIUM_CLIENT_SECRET"]
    )
    # Take the first complete audit from the list, then fetch it by id.
    for item in api.iter_list("/v2/audits", "audit_id", status="complete"):
        audit = api.request("GET", f"/v2/audits/{item['audit_id']}")
        print(
            audit["audit_id"],
            audit["status"],
            audit["score"],
            audit["set_id"],
            len(audit["documents"]),
        )
        break