Recipes

Keep audits in sync with updated_after

Use this when you keep a copy of audits in your own database, electronic health record (EHR) or warehouse. Each run pulls only what changed since the last run: new audits, completions, rescores, review marks and flag changes. A run costs one request per 100 changed audits. Fetching each stored audit instead would cost one request per audit.

Prerequisites

  • An access token in $BRELLIUM_ACCESS_TOKEN, cached and reused (Authenticate with OAuth2 client credentials).
  • A table keyed on audit_id and one stored checkpoint timestamp. The short samples keep the checkpoint in a file named last_pull, and the full programs in step 2 use an in-memory map, so every sample runs as it is.
  • With a live key the response contains protected health information (PHI); a sandbox key returns sample audits.

Steps

Backfill once

The first run has no updated_after, so it pages through every audit, newest first, up to the moment it started. Record that moment before the first request, and save it as the checkpoint only after the loop finishes. An audit updated during the backfill then falls into the first window, and if the backfill fails, the next run repeats it.

NOW="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
after=""
while :; do
  url="https://api.brellium.com/v2/audits?status=complete&limit=100"
  url="$url&updated_before=$NOW"
  if [ -n "$after" ]; then url="$url&starting_after=$after"; fi
  page=$(curl --silent "$url" \
    --header "Authorization: Bearer $BRELLIUM_ACCESS_TOKEN")
  # Upsert each row by audit_id here.
  echo "$page" | jq -c '.data[] | {audit_id, status, score, flags, updated_at}'
  if [ "$(echo "$page" | jq -r '.has_more')" != "true" ]; then break; fi
  after=$(echo "$page" | jq -r '.data[-1].audit_id')
done
echo "$NOW" > last_pull    # the checkpoint, written only after the loop finished

import { writeFile } from "node:fs/promises";

const now = new Date().toISOString();
let after = "";
while (true) {
  const params = new URLSearchParams({
    status: "complete",
    updated_before: now,
    limit: "100",
  });
  if (after) params.set("starting_after", after);
  const res = await fetch(`https://api.brellium.com/v2/audits?${params}`, {
    headers: { Authorization: `Bearer ${process.env.BRELLIUM_ACCESS_TOKEN}` },
  });
  if (!res.ok) {
    const { error } = await res.json();
    throw new Error(`${res.status} ${error.code}: ${error.message}`);
  }
  const page = await res.json();
  for (const { audit_id, status, score, flags, updated_at } of page.data) {
    // upsert each row by audit_id here
    console.log(JSON.stringify({ audit_id, status, score, flags, updated_at }));
  }
  if (!page.has_more) break;
  after = page.data.at(-1).audit_id;
}
// the checkpoint, written only after the loop finished
await writeFile("last_pull", now);

import { writeFile } from "node:fs/promises";
import BrelliumClient from "@brellium/api-client";

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

const now = new Date().toISOString();
let after: string | undefined;
while (true) {
  const page = await client.audits.list({
    status: "complete",
    updated_before: now,
    limit: 100,
    starting_after: after,
  });
  for (const { audit_id, status, score, flags, updated_at } of page.data) {
    // upsert each row by audit_id here
    console.log(JSON.stringify({ audit_id, status, score, flags, updated_at }));
  }
  if (!page.has_more) break;
  after = page.data[page.data.length - 1].audit_id;
}
// the checkpoint, written only after the loop finished
await writeFile("last_pull", now);

import json
import os
from datetime import datetime, timezone
from pathlib import Path

from brellium_api_client import BrelliumClient, omit

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

now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
after: str | None = None
while True:
    page = client.audits.list(
        status="complete",
        updated_before=now,
        limit=100,
        starting_after=after or omit,
    )
    for a in page.data:
        row = a.model_dump(
            mode="json",
            include={"audit_id", "status", "score", "flags", "updated_at"},
        )
        print(json.dumps(row))  # upsert each row by audit_id here
    if not page.has_more:
        break
    after = page.data[-1].audit_id
# the checkpoint, written only after the loop finished
Path("last_pull").write_text(now)

Each page has the shape shown in step 2. status=complete copies only finished audits. Leave it out to copy unfinished ones too. Add include_total=true to the first request for a total_count to drive a progress bar, and leave it off the other pages because counting makes the request slower.

Pull the next window

Lists can take a few seconds to include a new or changed audit, so an audit updated shortly before your run started can be missing from that run. Start each window ten minutes before the previous run began, which leaves far more margin than those few seconds. Both bounds are inclusive, so an audit updated exactly on a boundary can also appear in two runs. Because you upsert by audit_id, these repeats do no harm.

Each run works from these values:

now            = the time this run starts, RFC 3339 in UTC with a trailing Z
last_pull      = the previous run's `now`, unchanged if that run failed
updated_after  = last_pull minus 10 minutes
updated_before = now

updated_before = now keeps the window fixed while you page, so the next run picks up an audit that changes during this one. The scheduled version derives the bounds from the checkpoint and pages like the backfill:

LAST_PULL="$(cat last_pull)"
NOW="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
# Ten minutes before the previous run began.
AFTER="$(jq -rn --arg t "$LAST_PULL" '$t | fromdate - 600 | todate')"

after=""
while :; do
  url="https://api.brellium.com/v2/audits?status=complete&limit=100"
  url="$url&updated_after=$AFTER&updated_before=$NOW"
  if [ -n "$after" ]; then url="$url&starting_after=$after"; fi
  page=$(curl --silent "$url" \
    --header "Authorization: Bearer $BRELLIUM_ACCESS_TOKEN")
  # Upsert each row by audit_id here.
  echo "$page" | jq -c '.data[] | {audit_id, status, score, flags, updated_at}'
  if [ "$(echo "$page" | jq -r '.has_more')" != "true" ]; then break; fi
  after=$(echo "$page" | jq -r '.data[-1].audit_id')
done
echo "$NOW" > last_pull

import { readFile, writeFile } from "node:fs/promises";

const lastPull = (await readFile("last_pull", "utf8")).trim();
const now = new Date().toISOString();
// ten minutes before the previous run began
const updatedAfter = new Date(Date.parse(lastPull) - 600_000).toISOString();

let after = "";
while (true) {
  const params = new URLSearchParams({
    status: "complete",
    updated_after: updatedAfter,
    updated_before: now,
    limit: "100",
  });
  if (after) params.set("starting_after", after);
  const res = await fetch(`https://api.brellium.com/v2/audits?${params}`, {
    headers: { Authorization: `Bearer ${process.env.BRELLIUM_ACCESS_TOKEN}` },
  });
  if (!res.ok) {
    const { error } = await res.json();
    throw new Error(`${res.status} ${error.code}: ${error.message}`);
  }
  const page = await res.json();
  for (const { audit_id, status, score, flags, updated_at } of page.data) {
    // upsert each row by audit_id here
    console.log(JSON.stringify({ audit_id, status, score, flags, updated_at }));
  }
  if (!page.has_more) break;
  after = page.data.at(-1).audit_id;
}
await writeFile("last_pull", now);

import { readFile, writeFile } from "node:fs/promises";
import BrelliumClient from "@brellium/api-client";

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

const lastPull = (await readFile("last_pull", "utf8")).trim();
const now = new Date().toISOString();
// ten minutes before the previous run began
const updatedAfter = new Date(Date.parse(lastPull) - 600_000).toISOString();

let after: string | undefined;
while (true) {
  const page = await client.audits.list({
    status: "complete",
    updated_after: updatedAfter,
    updated_before: now,
    limit: 100,
    starting_after: after,
  });
  for (const { audit_id, status, score, flags, updated_at } of page.data) {
    // upsert each row by audit_id here
    console.log(JSON.stringify({ audit_id, status, score, flags, updated_at }));
  }
  if (!page.has_more) break;
  after = page.data[page.data.length - 1].audit_id;
}
await writeFile("last_pull", now);

import json
import os
from datetime import datetime, timedelta, timezone
from pathlib import Path

from brellium_api_client import BrelliumClient, omit

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

last_pull = datetime.fromisoformat(
    Path("last_pull").read_text().strip().replace("Z", "+00:00")
)
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
# ten minutes before the previous run began
updated_after = (last_pull - timedelta(minutes=10)).strftime("%Y-%m-%dT%H:%M:%SZ")

after: str | None = None
while True:
    page = client.audits.list(
        status="complete",
        updated_after=updated_after,
        updated_before=now,
        limit=100,
        starting_after=after or omit,
    )
    for a in page.data:
        row = a.model_dump(
            mode="json",
            include={"audit_id", "status", "score", "flags", "updated_at"},
        )
        print(json.dumps(row))  # upsert each row by audit_id here
    if not page.has_more:
        break
    after = page.data[-1].audit_id
Path("last_pull").write_text(now)

For the full audit, see Retrieve an audit by id. A page inside the window looks like this, trimmed to the fields the loop reads:

{
  "object": "list",
  "data": [
    {
      // the upsert key
      "audit_id": "a1b2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d",
      "status": "complete",
      "score": 0.6666666666666666,
      "flags": ["Failed", "Missing Signature"],
      // newer than your stored copy: write it
      "updated_at": "2026-09-24T02:37:12.105Z"
      // set_id, created_at, reviewed_at, display_info, documents, people and
      // responses omitted
    }
  ],
  // pass the last audit_id as starting_after
  "has_more": true
}

An empty window is the common case:

// nothing changed; still advance the checkpoint
{ "object": "list", "data": [], "has_more": false }

The full programs below run one sync: the backfill when no checkpoint exists, otherwise one overlapping window. They keep the audits and the checkpoint in memory, so replace store with your database.

// Keep credentials on the server; never in a browser or mobile app.
const token = process.env.BRELLIUM_ACCESS_TOKEN;
if (!token) throw new Error("BRELLIUM_ACCESS_TOKEN is not set");

const BASE_URL = "https://api.brellium.com";
const OVERLAP_MS = 10 * 60 * 1000;

type Audit = {
  audit_id: string;
  object: "audit";
  status: string;
  score: number | null;
  set_id: string;
  flags: string[];
  created_at: string;
  updated_at: string;
  reviewed_at: string | null;
};
type Page = { object: "list"; data: Audit[]; has_more: boolean };

// Replace with your database: one upsert keyed on audit_id, one stored checkpoint.
const store = {
  audits: new Map<string, Audit>(),
  lastPull: null as string | null,
  upsert(audit: Audit) {
    const current = this.audits.get(audit.audit_id);
    // already have this version
    if (current && current.updated_at >= audit.updated_at) return;
    this.audits.set(audit.audit_id, audit);
  },
};

async function* listAudits(params: Record<string, string>): AsyncGenerator<Audit> {
  let startingAfter: string | undefined;
  let retries = 0;
  while (true) {
    const url = new URL("/v2/audits", BASE_URL);
    for (const [key, value] of Object.entries(params))
      url.searchParams.set(key, value);
    url.searchParams.set("limit", "100");
    if (startingAfter) url.searchParams.set("starting_after", startingAfter);

    const res = await fetch(url, { headers: { Authorization: `Bearer ${token}` } });
    // At most five 429s in a row per page; then throw like any other error.
    if (res.status === 429 && retries < 5) {
      retries += 1;
      const seconds = Number(res.headers.get("Retry-After") ?? "1");
      await new Promise((resolve) => setTimeout(resolve, seconds * 1000));
      continue; // retry the same page
    }
    if (!res.ok) throw new Error(`GET /v2/audits returned ${res.status}`);
    retries = 0;

    const page = (await res.json()) as Page;
    for (const audit of page.data) yield audit;
    if (!page.has_more) return;
    startingAfter = page.data[page.data.length - 1].audit_id;
  }
}

// One run: the backfill when no checkpoint exists, otherwise one overlapping
// window.
async function sync(filters: Record<string, string> = {}): Promise<number> {
  const now = new Date().toISOString(); // RFC 3339 in UTC with a trailing Z
  const params: Record<string, string> = { ...filters, updated_before: now };
  if (store.lastPull) {
    params.updated_after = new Date(
      Date.parse(store.lastPull) - OVERLAP_MS,
    ).toISOString();
  }

  let count = 0;
  for await (const audit of listAudits(params)) {
    store.upsert(audit);
    count += 1;
  }
  // Only after the whole window succeeded; a thrown error leaves the checkpoint
  // unchanged.
  store.lastPull = now;
  return count;
}

async function main() {
  const count = await sync({ status: "complete" });
  console.log(`synced ${count} audits; checkpoint ${store.lastPull}`);
}

main();

# Keep credentials on the server; never in a browser or mobile app.
import os
import time
from datetime import datetime, timedelta, timezone

import requests

TOKEN = os.environ["BRELLIUM_ACCESS_TOKEN"]
BASE_URL = "https://api.brellium.com"
OVERLAP = timedelta(minutes=10)


def rfc3339(moment: datetime) -> str:
    """RFC 3339 in UTC with a trailing Z; an offset such as +02:00 returns 400."""
    return (
        moment.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
    )


class MemoryStore:
    """Replace with your database: one upsert keyed on audit_id, one stored
    checkpoint.
    """

    def __init__(self) -> None:
        self.audits: dict[str, dict] = {}
        self.last_pull: datetime | None = None

    def upsert(self, audit: dict) -> None:
        current = self.audits.get(audit["audit_id"])
        if current and current["updated_at"] >= audit["updated_at"]:
            return  # already have this version
        self.audits[audit["audit_id"]] = audit


def list_audits(session: requests.Session, **params):
    params = {"limit": 100, **params}
    retries = 0
    while True:
        res = session.get(f"{BASE_URL}/v2/audits", params=params, timeout=30)
        # At most five 429s in a row per page; then raise like any other error.
        if res.status_code == 429 and retries < 5:
            retries += 1
            time.sleep(int(res.headers.get("Retry-After", "1")))
            continue  # retry the same page
        res.raise_for_status()
        retries = 0
        page = res.json()
        yield from page["data"]
        if not page["has_more"]:
            return
        params["starting_after"] = page["data"][-1]["audit_id"]


def sync(session: requests.Session, store: MemoryStore, **filters) -> int:
    """One run: the backfill when no checkpoint exists, otherwise one overlapping
    window.
    """
    now = datetime.now(timezone.utc)
    params = {**filters, "updated_before": rfc3339(now)}
    if store.last_pull is not None:
        params["updated_after"] = rfc3339(store.last_pull - OVERLAP)

    count = 0
    for audit in list_audits(session, **params):
        store.upsert(audit)
        count += 1
    # Only after the whole window succeeded; an exception leaves the checkpoint
    # unchanged.
    store.last_pull = now
    return count


if __name__ == "__main__":
    session = requests.Session()
    session.headers["Authorization"] = f"Bearer {TOKEN}"
    store = MemoryStore()
    count = sync(session, store, status="complete")
    print(f"synced {count} audits; checkpoint {rfc3339(store.last_pull)}")

After a 429, both programs wait Retry-After seconds and retry the page, up to five times in a row. When a run throws, they leave the checkpoint as it was, so the next run repeats the window.

This example shows the overlap at work, with a run every five minutes on 2026-10-21 UTC. Each request is shown without status=complete, limit=100 and the date:

Time What happens Your request and what it returns
10:00:00 Run 1, the backfill, starts. updated_before=10:00:00Z, no updated_after: every audit.
10:05:00 Run 2 starts. updated_after=09:50:00Z&updated_before=10:05:00Z: nothing new.
10:09:58 Your PATCH sets flags on audit C. Its updated_at is 10:09:58, but lists don't include the change yet.
10:10:00 Run 3 starts. updated_after=09:55:00Z&updated_before=10:10:00Z leaves out audit C, because lists don't include the change yet.
10:15:00 Run 4 starts. updated_after=10:00:00Z&updated_before=10:15:00Z returns audit C, because 10:09:58 is inside the overlap.

Without the overlap, run 4 would start at updated_after=10:10:00Z and audit C would stay missing until its next change.

Upsert by audit_id

Write every audit with an upsert keyed on audit_id and keep the version with the newest updated_at. A plain insert breaks because:

  • The overlap returns the same audit in consecutive runs.
  • An audit updated several times inside one window arrives once, in its latest version.
  • A complete audit can change again. When it is recalculated in Brellium, its status is completing briefly, then complete again, possibly with a different score and flags.
  • Brellium can run an audit again. Until the audit completes again, its status is in_progress, or it is missing from lists.
  • When the audit is marked reviewed or unreviewed in Brellium, reviewed_at changes and updated_at moves forward with it.
  • Your own PATCH /v2/audits/{audit_id} sets updated_at too.

In every case audit_id and created_at stay the same and updated_at moves forward.

Schedule it, and re-run after an incident

Run sync on a timer, for example every five minutes. Your organization can send a burst of 100 requests, then 5 per second, shared by every key; the sandbox has a separate budget with the same limits.

If Brellium tells you about an incident that delayed updates, run the loop once with updated_after set to before the incident started, for example the previous day. The upsert absorbs the repeated audits.

Expected response

Every item in data[] is the full audit, with the 13 fields described in Understand an audit and in List audits. Step 2 shows a page trimmed to the fields the loop reads. documents[] holds document_id values for Retrieve a document (GET /v2/documents/{document_id}).

Common mistakes

Mistake What you see Fix
Syncing on created_after New audits arrive, but completions, rescores, review marks and flag changes never do created_at never changes. Use updated_after and updated_before.
A timestamp with an offset (2026-10-21T10:00:00+02:00) or a date only 400 VALIDATION_ERROR; error.details[0] has message: "Invalid ISO datetime" and path: ["updated_after"] Send RFC 3339 timestamps in UTC ending in Z.
Asking for limit=1000 400 VALIDATION_ERROR; error.details[0] has message: "Too big: expected number to be <=100" and path: ["limit"] limit is 1 to 100. Use limit=100 and follow has_more.

Next

Understand an audit

The audit object, its status lifecycle, flags, updated_at, and its links to other resources.

Understand an audit

Page through and filter lists

Cursor pagination, batch get by id, and each resource's filters.

Page through and filter lists

Handle rate limits and retries

Read the RateLimit headers, back off on 429, retry only safe calls, and tell a 403 from a 429.

Handle rate limits and retries
Previous
Update audit flags
Next
Invite and manage users