Data export - GrainQL

This example builds a Node.js script that pages through the events of a date range and writes them to a CSV file and a JSON file. The script waits between pages to stay inside the rate limit of your plan, retries on 429 after Retry-After seconds, and runs from cron. The Query API needs the Starter plan or higher and an API key with the Query API permission. Read Query API.

Files

File	Holds
`src/grain-api.ts`	The API client and the retry logic.
`src/export.ts`	The script. It pages through the events and writes both files.
`.env`	`GRAIN_TENANT_ID`, `GRAIN_API_KEY`, and the pace of your plan.

The script uses the global fetch of Node.js 18 or newer and has no runtime dependency.

mkdir grain-data-export && cd grain-data-export
npm init -y
npm install -D typescript tsx @types/node

Put the tenant alias, the key, and the pace in .env.

GRAIN_TENANT_ID=your-tenant-id
GRAIN_API_KEY=$GRAIN_API_KEY
GRAIN_REQUESTS_PER_MINUTE=60

GRAIN_REQUESTS_PER_MINUTE sets the pace. The limit is 120 requests per minute for the whole workspace, so 60 leaves room for other clients. Read the pace section below.

The API client

send makes one request. On 429 it waits Retry-After seconds and retries, up to MAX_RETRIES times. On a network error it waits with exponential backoff. On any other error status it throws at once, because a retry cannot change a 401, a 403, or a 404.

// src/grain-api.ts
const BASE_URL = 'https://queryapis.grainql.com/v1/api/query';
const MAX_RETRIES = 3;

export interface QueryRequest {
  event?: string;
  after?: string;
  before?: string;
  filterSet?: Array<{ property: string; comparison: string; value: unknown }>;
  pagination?: { offset: number; size: number };
}

export interface GrainEvent {
  eventName: string;
  userId: string;
  eventTs: string;
  properties: Record<string, unknown>;
  eventDate: string;
  insertId: string;
}

export class GrainApiError extends Error {
  constructor(public status: number, message: string) {
    super(message);
  }
}

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

async function send<T>(path: string, init: RequestInit): Promise<T> {
  let backoffMs = 1000;
  for (let attempt = 1; ; attempt++) {
    let response: Response;
    try {
      response = await fetch(`${BASE_URL}${path}`, init);
    } catch (error) {
      if (attempt > MAX_RETRIES) throw error;
      await sleep(backoffMs);
      backoffMs *= 2;
      continue;
    }
    if (response.ok) return response.json();

    const body = await response.json().catch(() => ({ error: response.statusText }));
    if (response.status === 429 && attempt <= MAX_RETRIES) {
      const retryAfter = Number(response.headers.get('Retry-After')) || 60;
      console.error(`429 received. Waiting ${retryAfter} seconds.`);
      await sleep(retryAfter * 1000);
      continue;
    }
    throw new GrainApiError(response.status, body.error);
  }
}

export function createGrainApi(apiKey: string, tenantId: string) {
  const headers = { 'Content-Type': 'application/json', 'X-API-Key': apiKey };
  return {
    queryEvents: (request: QueryRequest) =>
      send<GrainEvent[]>(`/${tenantId}`, {
        method: 'POST',
        headers,
        body: JSON.stringify(request),
      }),
    countEvents: (request: Omit<QueryRequest, 'pagination'>) =>
      send<{ count: number }>(`/count/${tenantId}`, {
        method: 'POST',
        headers,
        body: JSON.stringify(request),
      }).then((result) => result.count),
    listEventNames: () => send<string[]>(`/events/${tenantId}`, { headers }),
  };
}

The script

The script counts the events first, then reads pages of 1,000 with a growing offset. Between two requests it waits 60 / GRAIN_REQUESTS_PER_MINUTE seconds. The CSV has one column per event field and one column per property key that appears in the export.

// src/export.ts
import { writeFileSync } from 'node:fs';
import { createGrainApi, type GrainEvent, type QueryRequest } from './grain-api';

const api = createGrainApi(process.env.GRAIN_API_KEY!, process.env.GRAIN_TENANT_ID!);
const PAGE_SIZE = 1000;
const PACE_MS = (60 / Number(process.env.GRAIN_REQUESTS_PER_MINUTE ?? 60)) * 1000;

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

async function exportAllEvents(request: Omit<QueryRequest, 'pagination'>): Promise<GrainEvent[]> {
  const total = await api.countEvents(request);
  console.log(`${total} events to export`);
  const events: GrainEvent[] = [];
  for (let offset = 0; offset < total; offset += PAGE_SIZE) {
    await sleep(PACE_MS);
    const page = await api.queryEvents({ ...request, pagination: { offset, size: PAGE_SIZE } });
    events.push(...page);
    console.log(`${Math.min(offset + PAGE_SIZE, total)} of ${total}`);
  }
  return events;
}

function csvCell(value: unknown): string {
  if (value === null || value === undefined) return '';
  const text = typeof value === 'object' ? JSON.stringify(value) : String(value);
  return /[",\n]/.test(text) ? `"${text.replace(/"/g, '""')}"` : text;
}

function toCsv(events: GrainEvent[]): string {
  const fields = ['eventName', 'userId', 'eventTs', 'eventDate', 'insertId'] as const;
  const propertyKeys = [...new Set(events.flatMap((event) => Object.keys(event.properties ?? {})))];
  const header = [...fields, ...propertyKeys.map((key) => `properties.${key}`)];
  const rows = events.map((event) =>
    [...fields.map((field) => event[field]), ...propertyKeys.map((key) => event.properties?.[key])]
      .map(csvCell)
      .join(','),
  );
  return [header.join(','), ...rows].join('\n') + '\n';
}

function isoDate(daysAgo: number): string {
  const date = new Date();
  date.setUTCDate(date.getUTCDate() - daysAgo);
  return date.toISOString().slice(0, 10);
}

async function main() {
  const [after = isoDate(1), before = isoDate(1), event] = process.argv.slice(2);
  const request: Omit<QueryRequest, 'pagination'> = { after, before, ...(event && { event }) };
  const events = await exportAllEvents(request);
  const stem = `export-${after}-${before}`;

  writeFileSync(`${stem}.csv`, toCsv(events));
  writeFileSync(
    `${stem}.json`,
    JSON.stringify(
      { metadata: { totalEvents: events.length, exportDate: new Date().toISOString(), request }, events },
      null,
      2,
    ),
  );
  console.log(`Wrote ${stem}.csv and ${stem}.json`);
}

main().catch((error) => {
  console.error(error.message);
  process.exit(1);
});

Run the script with a start date, an end date, and an optional event name. Without arguments it exports yesterday.

npx tsx src/export.ts 2024-01-01 2024-01-31 purchase_completed

15247 events to export
1000 of 15247
2000 of 15247
...
15247 of 15247
Wrote export-2024-01-01-2024-01-31.csv and export-2024-01-01-2024-01-31.json

The JSON file holds the events with the shape that the Query API returns:

{
  "metadata": {
    "totalEvents": 15247,
    "exportDate": "2024-02-01T02:00:07.412Z",
    "request": { "after": "2024-01-01", "before": "2024-01-31", "event": "purchase_completed" }
  },
  "events": [
    {
      "eventName": "purchase_completed",
      "userId": "user_123",
      "eventTs": "2024-01-15T10:30:00Z",
      "properties": { "price": 149.0, "currency": "USD" },
      "eventDate": "2024-01-15",
      "insertId": "unique-event-id"
    }
  ]
}

To export one segment, add a filterSet to request in main. Read Filters.

const request = {
  after,
  before,
  event: 'purchase_completed',
  filterSet: [{ property: 'properties.price', comparison: 'GREATER_THAN', value: 100 }],
};

Pace and budget

An export of N events costs 1 + ceil(N / 1000) requests: one count and one request per page. The rate limit is 120 requests per minute for the whole workspace, on every plan. There is no daily limit.

`GRAIN_REQUESTS_PER_MINUTE`	Wait between pages	Events per hour
`60` (recommended)	1 s	about 3,600,000
`120` (the limit)	0.5 s	about 7,200,000

At 120, every other client of the workspace gets 429 while the export runs. Keep the value at 60 or lower.

Scheduled run

Run the script from cron at a fixed time. The line below exports yesterday every day at 02:00 and appends the log to export.log.

crontab -e

0 2 * * * cd /srv/grain-data-export && npx tsx src/export.ts >> export.log 2>&1

Rules for the schedule:

Load .env in the cron environment. Cron does not read your shell profile. Use node --env-file=.env (Node.js 20.6 or newer) or export the variables in the crontab.
Leave a gap between two scheduled exports of one workspace. Both count against one limit.
Make sure that a run finishes before the next one starts. An export of 100,000 events takes about 2 minutes at 1 second per page.

Errors

Status	Body	What to do
`401`	`{"error":"Missing X-API-Key header"}` or `{"error":"Invalid API key or insufficient permissions"}`	Make sure that `GRAIN_API_KEY` is set and that the key has the Query API permission.
`403`	`{"error":"Query API requires Starter plan or higher. ..."}`	Choose a plan for the workspace. The script stops.
`404`	`{"error":"Tenant not found"}`	Use the tenant alias from the dashboard in `GRAIN_TENANT_ID`, not the UUID.
`429`	`{"error":"rate_limit_exceeded", ...}` with `Retry-After`	The client waits `Retry-After` seconds and retries up to 3 times. If it still fails, lower `GRAIN_REQUESTS_PER_MINUTE` or run the export later.
`400`	`{"error":"Invalid date format. Use YYYY-MM-DD format."}`	Pass dates as `YYYY-MM-DD`.

Limits

Limit	Value	Where
Events per page	1,000	`pagination.size`. `PAGE_SIZE` in `src/export.ts`.
Retries per request	3	`MAX_RETRIES` in `src/grain-api.ts`.
Requests per minute	120	Per workspace, on every plan. Every key of the workspace shares it.
Memory	All events of one run	The script holds the events in memory before it writes the files. Export one day per run.

Read Custom dashboard for a Next.js page that polls the same endpoints.

​Files

​The API client

​The script

​Pace and budget

​Scheduled run

​Errors

​Limits