Warehouse sync

The job

Your analytics team wants every fileAI document, with its status and classification, in a raw_files table that refreshes every 15 minutes. Each run should fetch only what changed since the last one, and never lose or double-count a file.

Sample input

A workspace with about 40,000 files, of which a few hundred change between runs: new uploads, re-processed files, edited values. A small sync_state table in your warehouse holds one row: the watermark from the last successful run, 2026-10-15T23:45:00.000Z.

Steps

The cURL commands use two environment variables:

export FILEAI_API_KEY="your-api-key"
export FILEAI_BASE_URL="https://api.orion.file.ai/prod/v1"   # or your regional host

See Regions for regional hosts.

Read the watermark

Load the last saved updatedAt from sync_state. On the first run there is none: omit updatedAfter and load everything.

Walk every changed file with a cursor

Call GET /v1/files with updatedAfter, sorted by updatedAt ascending, in cursor mode. Send cursor empty on the first page, then follow pagination.nextCursor until hasMore is false. Keep every other parameter identical for the whole walk: a cursor is bound to its filters and sort, and changing them returns 400.

curl "$FILEAI_BASE_URL/files?updatedAfter=2026-10-15T23:45:00.000Z&sortBy=updatedAt&sortOrder=ASC&limit=100&cursor=" \
  -H "x-api-key: $FILEAI_API_KEY"

# next page: same query, cursor=<nextCursor>

import os, requests

BASE = "https://api.orion.file.ai/prod/v1"
HEADERS = {"x-api-key": os.environ["FILEAI_API_KEY"]}

def fetch_changed(watermark):
    params = {"sortBy": "updatedAt", "sortOrder": "ASC", "limit": 100, "cursor": ""}
    if watermark:
        params["updatedAfter"] = watermark
    rows = []
    while True:
        r = requests.get(f"{BASE}/files", headers=HEADERS, params=params)
        r.raise_for_status()
        page = r.json()
        rows.extend(page["data"])
        if not page["pagination"]["hasMore"]:
            return rows
        params["cursor"] = page["pagination"]["nextCursor"]

const BASE = "https://api.orion.file.ai/prod/v1";
const headers = { "x-api-key": process.env.FILEAI_API_KEY };

async function fetchChanged(watermark) {
  const params = new URLSearchParams({ sortBy: "updatedAt", sortOrder: "ASC", limit: "100", cursor: "" });
  if (watermark) params.set("updatedAfter", watermark);
  const rows = [];
  while (true) {
    const res = await fetch(`${BASE}/files?${params}`, { headers });
    if (!res.ok) throw new Error(`HTTP ${res.status}, X-Request-Id ${res.headers.get("x-request-id")}`);
    const page = await res.json();
    rows.push(...page.data);
    if (!page.pagination.hasMore) return rows;
    params.set("cursor", page.pagination.nextCursor);
  }
}

If the walk fails part-way, start again from the same watermark with an empty cursor. Nothing is lost, because the watermark has not moved.

Deduplicate and merge on fileId

updatedAfter is inclusive (>=), so the file at the watermark comes back again, and a file edited during the walk can appear twice. Stage the rows, keep the newest updatedAt per fileId, and merge. In Snowflake:

Snowflake

MERGE INTO raw_files t
USING (
  SELECT * FROM staging_files
  QUALIFY ROW_NUMBER() OVER (PARTITION BY file_id ORDER BY updated_at DESC) = 1
) s
ON t.file_id = s.file_id
WHEN MATCHED AND s.updated_at > t.updated_at THEN
  UPDATE SET file_name = s.file_name, status = s.status, file_class = s.file_class, updated_at = s.updated_at
WHEN NOT MATCHED THEN
  INSERT (file_id, file_name, status, file_class, updated_at)
  VALUES (s.file_id, s.file_name, s.status, s.file_class, s.updated_at);

Advance the watermark

Only after the merge commits, save the largest updatedAt you loaded. Keep the old value if the run returned nothing.

Python

rows = fetch_changed(watermark)
load_and_merge(rows)
new_watermark = max((r["updatedAt"] for r in rows), default=watermark)
save_watermark(new_watermark)

updatedAt is an ISO 8601 UTC string with millisecond precision, so the largest string is also the latest time.

Complete job

The steps above as one script you can schedule. It assumes three tables in your warehouse and a DB-API connection conn with autocommit off (for example from snowflake.connector.connect(...)).

Snowflake tables

CREATE TABLE IF NOT EXISTS raw_files (
  file_id    STRING PRIMARY KEY,
  file_name  STRING,
  status     STRING,
  file_class STRING,
  updated_at TIMESTAMP_NTZ
);
CREATE TABLE IF NOT EXISTS staging_files LIKE raw_files;
CREATE TABLE IF NOT EXISTS sync_state (name STRING PRIMARY KEY, watermark STRING);

sync_files.py

import os, time, requests

BASE = "https://api.orion.file.ai/prod/v1"   # or your regional host
HEADERS = {"x-api-key": os.environ["FILEAI_API_KEY"]}

MERGE_SQL = """
MERGE INTO raw_files t
USING (
  SELECT * FROM staging_files
  QUALIFY ROW_NUMBER() OVER (PARTITION BY file_id ORDER BY updated_at DESC) = 1
) s
ON t.file_id = s.file_id
WHEN MATCHED AND s.updated_at > t.updated_at THEN
  UPDATE SET file_name = s.file_name, status = s.status, file_class = s.file_class, updated_at = s.updated_at
WHEN NOT MATCHED THEN
  INSERT (file_id, file_name, status, file_class, updated_at)
  VALUES (s.file_id, s.file_name, s.status, s.file_class, s.updated_at)
"""


def get(path, params):
    """GET that waits out a 429 using the Retry-After header."""
    while True:
        r = requests.get(f"{BASE}{path}", headers=HEADERS, params=params)
        if r.status_code == 429:
            time.sleep(int(r.headers.get("Retry-After", "5")))
            continue
        if not r.ok:
            raise RuntimeError(f"HTTP {r.status_code}, X-Request-Id {r.headers.get('X-Request-Id')}: {r.text}")
        return r.json()


def fetch_changed(watermark):
    params = {"sortBy": "updatedAt", "sortOrder": "ASC", "limit": 100, "cursor": ""}
    if watermark:
        params["updatedAfter"] = watermark
    rows = []
    while True:
        page = get("/files", params)
        rows.extend(page["data"])
        if not page["pagination"]["hasMore"]:
            return rows
        params["cursor"] = page["pagination"]["nextCursor"]


def read_watermark(cur):
    cur.execute("SELECT watermark FROM sync_state WHERE name = 'files'")
    row = cur.fetchone()
    return row[0] if row else None


def sync(conn):
    cur = conn.cursor()
    watermark = read_watermark(cur)
    rows = fetch_changed(watermark)
    if not rows:
        return watermark

    cur.execute("DELETE FROM staging_files")
    cur.executemany(
        "INSERT INTO staging_files (file_id, file_name, status, file_class, updated_at) "
        "VALUES (%s, %s, %s, %s, %s)",
        [(r["fileId"], r["fileName"], r["status"], r.get("fileClass"), r["updatedAt"]) for r in rows],
    )
    cur.execute(MERGE_SQL)

    new_watermark = max(r["updatedAt"] for r in rows)
    cur.execute(
        "MERGE INTO sync_state t USING (SELECT 'files' AS name, %s AS watermark) s ON t.name = s.name "
        "WHEN MATCHED THEN UPDATE SET watermark = s.watermark "
        "WHEN NOT MATCHED THEN INSERT (name, watermark) VALUES (s.name, s.watermark)",
        (new_watermark,),
    )
    conn.commit()   # the merge and the new watermark land together, or not at all
    return new_watermark

Run sync(conn) every 15 minutes from your scheduler. If it raises, roll back: nothing is committed, and the next run starts again from the same watermark.

Remove deleted files

Deleted files never appear in GET /v1/files, so an incremental run can’t see them. Once a day, or as often as you need, walk every file with no updatedAfter and delete warehouse rows that didn’t come back:

Python

def remove_deleted(conn):
    live = {r["fileId"] for r in fetch_changed(None)}
    cur = conn.cursor()
    cur.execute("SELECT file_id FROM raw_files")
    gone = [(fid,) for (fid,) in cur.fetchall() if fid not in live]
    if gone:
        cur.executemany("DELETE FROM raw_files WHERE file_id = %s", gone)
    conn.commit()

A full walk of a large workspace makes many requests; pace it with the rate-limit headers.

Result

Illustrative: shape from the API reference. Rows are shortened to the fields this job loads.

GET /v1/files?updatedAfter=…&sortBy=updatedAt&sortOrder=ASC&limit=100&cursor=

{
  "data": [
    {
      "fileId": "b765bd00-181e-4737-b4d9-004dadd6bd45",
      "uploadId": "e9818d17-e5be-4d57-bdc8-5a40b4f6f4e1",
      "fileName": "inv-2026-0412.pdf",
      "status": "processed",
      "fileClass": "Invoice",
      "schemaId": "6835aca6c2b1f40012ab3301",
      "createdAt": "2026-10-15T23:40:12.310Z",
      "updatedAt": "2026-10-15T23:45:00.000Z"
    },
    {
      "fileId": "3c1a9f02-77e4-4b1e-9a51-2f0d6c8e4a10",
      "uploadId": "f2538513-f0b9-4aa8-9c57-bc0a85c77de6",
      "fileName": "statement-2026-09.pdf",
      "status": "action_required",
      "fileClass": "Bank Statement",
      "schemaId": "6835aca6c2b1f40012ab3302",
      "createdAt": "2026-10-15T23:51:40.002Z",
      "updatedAt": "2026-10-15T23:58:21.744Z"
    }
  ],
  "pagination": {
    "limit": 100,
    "nextCursor": null,
    "hasMore": false
  }
}

The first row has updatedAt equal to the watermark: it was loaded last run and comes back because the filter is inclusive. The merge leaves it unchanged. The new watermark is 2026-10-15T23:58:21.744Z.

What it shows

A stored watermark plus updatedAfter keeps each run small, whatever the size of the workspace.
Cursor mode keeps the walk stable while files change. See Pagination for why legacy page/limit is a poor fit for syncs.
Inclusive filtering plus deduplication on fileId makes overlap between runs harmless.
Advancing the watermark last makes a failed run safe to repeat.

Deleted files never appear in GET /v1/files; use Remove deleted files to clear them from your warehouse. To load extracted values as well, call GET /v1/files/{fileId}/values for each changed fileId after the merge and read formValues from the response. Number cells whose value is 0 come back as null. See the financial report example and Data types.

Next steps

Incremental loading

Filters, what changes updatedAt, and every file field.

Pagination

Cursor rules, limits and the legacy mode.

Rate limits

Pace a large backfill and handle 429.

Troubleshooting

Cursor 400s and other common failures.

Get started

Products

Build

Go live

The job

Sample input

Steps

Complete job

Remove deleted files

Result

What it shows

Next steps

Incremental loading

Pagination

Rate limits

Troubleshooting

​The job

​Sample input

​Steps

​Complete job

​Remove deleted files

​Result

​What it shows

​Next steps

Incremental loading

Pagination

Rate limits

Troubleshooting

The job

Sample input

Steps

Complete job

Remove deleted files

Result

What it shows

Next steps