Постраничная навигация

set -euo pipefail
# US: BASE="https://api.us-east-1.livingsecurity.ai/v1"
BASE="https://api.eu-west-1.livingsecurity.ai/v1"
: "${LIVING_SECURITY_API_KEY:?Set LIVING_SECURITY_API_KEY}"
TOKEN=""

# Bound each run. Increase this page budget deliberately for larger collections.
for ((page_number=1; page_number<=1000; page_number++)); do
  PARAMS=(--data-urlencode "limit=500" --data-urlencode "type=person")
  if [ -n "$TOKEN" ]; then
    PARAMS+=(--data-urlencode "page_token=$TOKEN")
  fi
  RESULT=$(curl --silent --show-error --fail-with-body \
    --connect-timeout 5 --max-time 30 --get \
    -u "apikey:$LIVING_SECURITY_API_KEY" \
    -H "Living-Security-Version: 2026-07-01" \
    --write-out '\n%{http_code}' "${PARAMS[@]}" "$BASE/entities")
  STATUS="${RESULT##*$'\n'}"
  PAGE="${RESULT%$'\n'*}"
  [ "$STATUS" = "200" ] || { echo "Unexpected HTTP status: $STATUS" >&2; exit 1; }
  jq -e 'has("next_page_token") and .object == "list" and (.data | type == "array") and
    ((.has_more == false and .next_page_token == null) or
     (.has_more == true and (.next_page_token | type == "string" and length > 0)))' \
    <<< "$PAGE" > /dev/null
  NEXT=$(jq -r '.next_page_token // empty' <<< "$PAGE")
  [ -z "$NEXT" ] || [ "$NEXT" != "$TOKEN" ] || { echo "Repeated page token" >&2; exit 1; }
  jq -r '.data[].id' <<< "$PAGE"
  [ -n "$NEXT" ] || exit 0
  TOKEN="$NEXT"
done
echo "Page budget exhausted; the collection is incomplete" >&2
exit 1

import os
import requests

# US: BASE = "https://api.us-east-1.livingsecurity.ai/v1"
BASE = "https://api.eu-west-1.livingsecurity.ai/v1"
AUTH = ("apikey", os.environ["LIVING_SECURITY_API_KEY"])

def iter_entities(*, entity_type="person", max_pages=1000):
    if type(max_pages) is not int or max_pages < 1:
        raise ValueError("max_pages must be a positive integer")
    params = {"limit": 500, "type": entity_type}
    for _ in range(max_pages):
        response = requests.get(
            f"{BASE}/entities",
            auth=AUTH,
            params=params,
            headers={"Living-Security-Version": "2026-07-01"},
            timeout=(5, 30),
            allow_redirects=False,
        )
        try:
            response.raise_for_status()
            if response.status_code != 200:
                raise RuntimeError("Unexpected response; check your regional API URL")
            page = response.json()
        finally:
            response.close()
        if (not isinstance(page, dict) or page.get("object") != "list"
                or not isinstance(page.get("data"), list)
                or type(page.get("has_more")) is not bool
                or "next_page_token" not in page):
            raise RuntimeError("Invalid list response")
        token = page["next_page_token"]
        if ((page["has_more"] and (not isinstance(token, str) or not token))
                or (not page["has_more"] and token is not None)):
            raise RuntimeError("Inconsistent list response")
        if token is not None and token == params.get("page_token"):
            raise RuntimeError("Repeated page token")
        yield from page["data"]
        if token is None:
            return
        params = {**params, "page_token": token}
    raise RuntimeError("Page budget exhausted; the collection is incomplete")

if __name__ == "__main__":
    for entity in iter_entities():
        print(entity["id"])