# Warehouse sync
Source: https://docs.file.ai/examples/warehouse-sync

> Load new and changed files into your warehouse on a schedule, with a stored watermark, cursor pagination and deduplication.

Examples

# Warehouse sync

Copy pageCopy page

Load new and changed files into your warehouse on a schedule, with a stored watermark, cursor pagination and deduplication.

Copy pageCopy page

## 

[​

](#the-job)

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)

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)

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](/authentication#regions) for regional hosts.

1

Read the watermark

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

2

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

Python

Node.js

```
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.

3

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);
```

4

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)

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)

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](/api/rate-limits).

## 

[​

](#result)

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)

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](/api/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](#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](/examples/financial-report#result) and [Data types](/extract/data-types).

## 

[​

](#next-steps)

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 `400`s and other common failures.

[Previous](/examples/bank-statement-extraction)[

Minto

Hold a conversation with the fileAI agent over your workspace documents, one async turn at a time.

Next

](/ask/agent-sessions)

⌘I

## On this page

*   [The job](#the-job)
*   [Sample input](#sample-input)
*   [Steps](#steps)
*   [Complete job](#complete-job)
    *   [Remove deleted files](#remove-deleted-files)
*   [Result](#result)
*   [What it shows](#what-it-shows)
*   [Next steps](#next-steps)