Pagination & Delta Queries

Query API responses can contain thousands of records. Use pagination to retrieve data in manageable chunks, and delta queries to fetch only new data since your last sync.

Cursor-based pagination

Every response includes a next_cursor for fetching the next page:

{
  "data": [...],
  "count": 100,
  "has_more": true,
  "next_cursor": "eyJsYXN0X2lkIjogMTIzNH0="
}

Pass the cursor to get the next page:

async function fetchAllSearchHistory(uid: string) {
  const allData: SearchEntry[] = [];
  let cursor: string | undefined;

  do {
    const params = new URLSearchParams({ uid });
    if (cursor) {
      params.set('cursor', cursor);
    }

    const response = await fetch(
      `https://query.emergedata.ai/v1/sync/get_search?${params}`,
      { headers: { 'Authorization': `Bearer ${API_TOKEN}` } }
    );

    const result = await response.json();
    allData.push(...result.data);

    cursor = result.has_more ? result.next_cursor : undefined;
  } while (cursor);

  return allData;
}

def fetch_all_search_history(uid: str) -> list[dict]:
    all_data = []
    cursor = None

    while True:
        params = {'uid': uid}
        if cursor:
            params['cursor'] = cursor

        response = requests.get(
            'https://query.emergedata.ai/v1/sync/get_search',
            params=params,
            headers={'Authorization': f'Bearer {API_TOKEN}'}
        )

        result = response.json()
        all_data.extend(result['data'])

        if not result.get('has_more'):
            break
        cursor = result['next_cursor']

    return all_data

Page size

Control the number of records per page with the limit parameter:

curl 'https://query.emergedata.ai/v1/sync/get_search?uid=psub_c3d4e5f6789012345678901234abcdef&limit=500' \
  -H 'Authorization: Bearer your_api_token'

Parameter	Default	Maximum
`limit`	100	1000

Delta queries

Fetch only data that’s been ingested since your last sync using ingested_begin and ingested_end:

curl 'https://query.emergedata.ai/v1/sync/get_search?uid=psub_c3d4e5f6789012345678901234abcdef&ingested_begin=2024-01-15T00:00:00Z' \
  -H 'Authorization: Bearer your_api_token'

Time-based filtering

Parameter	Description	Format
`ingested_begin`	Records ingested at or after this time	ISO 8601
`ingested_end`	Records ingested before this time	ISO 8601

ingested_begin and ingested_end filter by when data was added to Emerge, not when the user performed the action. This is useful for incremental syncs.

Response fields for delta sync

The response includes additional fields to help with delta syncing:

{
  "data": [...],
  "count": 50,
  "has_more": false,
  "next_cursor": null,
  "applied_ingested_end": "2024-01-15T12:00:00Z"
}

Field	Description
`applied_ingested_end`	The actual end time used for filtering. Use this as your next `ingested_begin` value to ensure no gaps.

Delta sync pattern

interface SyncState {
  lastIngestedEnd: string;
}

async function deltaSync(uid: string, state: SyncState): Promise<SyncState> {
  const newData: SearchEntry[] = [];
  let cursor: string | undefined;

  do {
    const params = new URLSearchParams({
      uid,
      ingested_begin: state.lastIngestedEnd
    });
    if (cursor) {
      params.set('cursor', cursor);
    }

    const response = await fetch(
      `https://query.emergedata.ai/v1/sync/get_search?${params}`,
      { headers: { 'Authorization': `Bearer ${API_TOKEN}` } }
    );

    const result = await response.json();
    newData.push(...result.data);

    // Use applied_ingested_end for next sync to avoid gaps
    const appliedEnd = result.applied_ingested_end;

    cursor = result.has_more ? result.next_cursor : undefined;

    // After last page, update state
    if (!cursor) {
      return {
        lastIngestedEnd: appliedEnd
      };
    }
  } while (cursor);

  // Process new data
  await processNewData(newData);

  return state;
}

// Usage
let syncState: SyncState = {
  lastIngestedEnd: '2024-01-01T00:00:00Z'
};

// Run periodically
syncState = await deltaSync('psub_c3d4e5f6789012345678901234abcdef', syncState);

from dataclasses import dataclass
from typing import Optional

@dataclass
class SyncState:
    last_ingested_end: str


def delta_sync(uid: str, state: SyncState) -> SyncState:
    new_data = []
    cursor = None
    applied_end = None

    while True:
        params = {
            'uid': uid,
            'ingested_begin': state.last_ingested_end
        }
        if cursor:
            params['cursor'] = cursor

        response = requests.get(
            'https://query.emergedata.ai/v1/sync/get_search',
            params=params,
            headers={'Authorization': f'Bearer {API_TOKEN}'}
        )

        result = response.json()
        new_data.extend(result['data'])

        # Track applied_ingested_end for next sync
        applied_end = result.get('applied_ingested_end')

        if not result.get('has_more'):
            break
        cursor = result['next_cursor']

    # Process new data
    process_new_data(new_data)

    # Return new state with applied_ingested_end
    return SyncState(last_ingested_end=applied_end or state.last_ingested_end)


# Usage
sync_state = SyncState(last_ingested_end='2024-01-01T00:00:00Z')

# Run periodically
sync_state = delta_sync('psub_c3d4e5f6789012345678901234abcdef', sync_state)

Best practices

Use applied_ingested_end for state

Always use the applied_ingested_end from the response as your next ingested_begin value. This ensures no gaps between syncs.

Handle empty results

An empty data array with has_more: false means no new data since your last sync.

Use reasonable page sizes

500 records per page balances throughput and memory usage for most applications.

Implement backoff on errors

If you hit rate limits or errors, back off exponentially before retrying.

Deduplicate with event_id

Each record includes an event_id for deduplication. Use this to handle any potential duplicates across pagination boundaries.

Async job pagination

For async queries, results are returned as a single Parquet file. The file contains all records for all users in the request. If you need to paginate through very large async results, use the begin and end date parameters to narrow the time range:

// First batch: January 1-15
const firstWindow = new URLSearchParams({
  user_ids: 'psub_a1b2c3d4e5f6789012345678901234ab',
  begin: '2024-01-01T00:00:00Z',
  end: '2024-01-15T23:59:59Z'
});

const job1 = await fetch(
  `https://query.emergedata.ai/v1/search?${firstWindow.toString()}`,
  { headers: { 'Authorization': `Bearer ${API_TOKEN}` } }
).then(r => r.json());

// Second batch: January 16-31
const secondWindow = new URLSearchParams({
  user_ids: 'psub_a1b2c3d4e5f6789012345678901234ab',
  begin: '2024-01-16T00:00:00Z',
  end: '2024-01-31T23:59:59Z'
});

const job2 = await fetch(
  `https://query.emergedata.ai/v1/search?${secondWindow.toString()}`,
  { headers: { 'Authorization': `Bearer ${API_TOKEN}` } }
).then(r => r.json());

Guides

Async Endpoints

Sync Endpoints

Pagination & Delta Queries

Page size

Delta queries

Time-based filtering

Response fields for delta sync

Delta sync pattern

Best practices

​Cursor-based pagination

​Page size

​Delta queries

​Time-based filtering

​Response fields for delta sync

​Delta sync pattern

​Best practices

​Async job pagination

Cursor-based pagination

Page size

Delta queries

Time-based filtering

Response fields for delta sync

Delta sync pattern

Best practices

Async job pagination