Integration recipes

# Sex offender screening — Python 3.11+, requests. Env: OFFENDERSEARCH_API_KEY, OFFENDERSEARCH_BASE
import os, time, requests

API_KEY = os.environ["OFFENDERSEARCH_API_KEY"]
BASE = os.environ.get("OFFENDERSEARCH_BASE", "https://api.offendersearch.app")
H = {"X-API-Key": API_KEY, "Content-Type": "application/json"}

def search(first: str, last: str, dob: str | None = None, state: str | None = None) -> dict:
    body = {"query": {"firstName": first, "lastName": last, **({"dob": dob} if dob else {})}}
    if state:
        body["jurisdictions"] = [state]
    for attempt in range(4):
        r = requests.post(f"{BASE}/v1/search", headers=H, json=body, timeout=60)
        if r.status_code == 429:
            time.sleep(int(r.headers.get("Retry-After", "5"))); continue
        if r.status_code >= 500:
            time.sleep(2 ** attempt); continue
        r.raise_for_status()
        return r.json()
    raise RuntimeError("unavailable")

def classify(data: dict) -> str:
    if not data["records"]:
        return "clear" if data["status"] == "complete" else "incomplete"
    strong = [x for x in data["records"] if x.get("matchState") == "dob_match"]  # DOB corroborated; gate on matchState, not on matchConfidence alone
    return "match" if strong else "review"

if __name__ == "__main__":
    data = search("Jordan", "Rivera", "1988-04-12")
    print(classify(data), data["searchId"], [s["source"] for s in data["sourceStatus"] if s["incomplete"]])

// Criminal search in a Node.js service — Node 18+. Env: OFFENDERSEARCH_API_KEY, OFFENDERSEARCH_BASE
const API_KEY = process.env.OFFENDERSEARCH_API_KEY;
const BASE = process.env.OFFENDERSEARCH_BASE ?? "https://api.offendersearch.app";
const H = { "X-API-Key": API_KEY, "Content-Type": "application/json" };
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

export async function criminalSearch({ firstName, lastName, dob, state, page = 1 }) {
  const body = { query: { lastName, firstName, ...(dob ? { dob } : {}), ...(state ? { state } : {}) }, page, perPage: 50 };
  let r = await fetch(`${BASE}/v1/criminal/search`, { method: "POST", headers: H, body: JSON.stringify(body) });
  if (r.status === 429) { await sleep(1000 * Number(r.headers.get("Retry-After") ?? 5)); return criminalSearch({ firstName, lastName, dob, state, page }); }
  if (r.status === 402) throw new Error("trial used up or no active subscription");   // body: { detail }
  if (r.status === 422) throw new Error(`refused: ${JSON.stringify((await r.json()).error)}`);  // { error: { code, message } }
  if (r.status === 202) {                      // more than two live sources: the job runs in the background
    const { poll } = await r.json();           // "/v1/criminal/searches/crs_…"
    for (let i = 0; i < 150; i++) {            // every 2 s, up to 5 minutes
      await sleep(2000);
      const p = await fetch(`${BASE}${poll}`, { headers: H });
      if (p.status === 202) continue;          // JobRecord, status "pending", result null
      if (p.status === 422) throw new Error("refused when run: the stored query was too broad");
      if (!p.ok) throw new Error(`poll HTTP ${p.status}`);
      return (await p.json()).result;          // 200: JobRecord — the envelope is under result
    }
    throw new Error("live check did not finish");
  }
  if (!r.ok) throw new Error(`HTTP ${r.status}`);
  return r.json();
}

export function summarise(data) {
  // Each record: recordType, currentStatus, cases[].charges[] with description/statute/severity/disposition.
  return data.records.map((rec) => ({
    type: rec.recordType,
    status: rec.currentStatus,
    confidence: rec.matchConfidence,
    charges: (rec.cases ?? []).flatMap((c) => (c.charges ?? []).map((ch) => ({
      description: ch.description, statute: ch.statute, severity: ch.severity, disposition: ch.disposition,
    }))),
    source: rec.sources?.[0]?.code,
  }));
}

# screening/services.py — Django 4+/5. Settings: OFFENDERSEARCH_API_KEY, OFFENDERSEARCH_BASE
import time, requests
from django.conf import settings

H = {"X-API-Key": settings.OFFENDERSEARCH_API_KEY, "Content-Type": "application/json"}
BASE = getattr(settings, "OFFENDERSEARCH_BASE", "https://api.offendersearch.app")

class ScreeningError(Exception):
    pass

def criminal_search(*, last_name: str, first_name: str = "", dob: str | None = None, state: str | None = None) -> dict:
    query = {"lastName": last_name}
    if first_name: query["firstName"] = first_name
    if dob: query["dob"] = dob
    if state: query["state"] = state
    for attempt in range(4):
        r = requests.post(f"{BASE}/v1/criminal/search", headers=H, json={"query": query, "perPage": 50}, timeout=60)
        if r.status_code == 429:
            time.sleep(int(r.headers.get("Retry-After", "5"))); continue
        if r.status_code >= 500:
            time.sleep(2 ** attempt); continue
        if r.status_code == 402: raise ScreeningError("trial used up or no active subscription")  # body: {"detail": ...}
        if r.status_code == 422: raise ScreeningError(f"refused: {r.json()['error']}")  # {"error": {"code", "message"}}
        if r.status_code == 202:  # more than two live sources: the job runs in the background

# Create a person watch delivered by webhook (also by email if you add channels.email)
curl -X POST https://api.offendersearch.app/v1/monitors \
  -H "X-API-Key: $OFFENDERSEARCH_API_KEY" -H "Content-Type: application/json" \
  -d '{
    "product": "sex-offender",
    "type": "person",
    "label": "Jordan Rivera — resident 48213",
    "person": { "firstName": "Jordan", "lastName": "Rivera", "dob": "1988-04-12" },
    "channels": { "webhookUrl": "https://example.com/hooks/offendersearch?token=YOUR_SECRET" }
  }'

# A monitor watches one dataset: repeat with "product": "criminal" to watch criminal records too.

# Read alert history
curl https://api.offendersearch.app/v1/monitors/MONITOR_ID/alerts -H "X-API-Key: $OFFENDERSEARCH_API_KEY"