Class RecordsAPI

Aggregate records from a stream

Aggregate data for records from a stream.

const response = await client.records.aggregate('my_stream', {
  aggregates: {
    total_count: { count: {} },
    by_category: {
      uniqueValues: {
        property: ['mySpace', 'myContainer', 'category'],
        aggregates: { total: { sum: { property: ['mySpace', 'myContainer', 'amount'] } } }
      }
    }
  }
});
Copy

Delete records from a stream

Delete records from a mutable stream. The operation is idempotent - deleting non-existent records will not cause an error.

Note: This endpoint is only available for mutable streams.

await client.records.delete('my_mutable_stream', [
  { space: 'mySpace', externalId: 'record1' },
  { space: 'mySpace', externalId: 'record2' }
]);
Copy

Filter records from a stream

Retrieve records from a stream using filters.

const response = await client.records.filter('my_stream', {
  sources: [
    {
      source: { type: 'container', space: 'mySpace', externalId: 'myContainer' },
      properties: ['*']
    }
  ],
  filter: {
    equals: {
      property: ['mySpace', 'myContainer', 'status'],
      value: 'active'
    }
  },
  limit: 100
});
Copy

Ingest records into a stream

Ingest records into a stream.

await client.records.ingest('my_stream', [
  {
    space: 'mySpace',
    externalId: 'record1',
    sources: [
      {
        source: { type: 'container', space: 'mySpace', externalId: 'myContainer' },
        properties: { temperature: 25.5, timestamp: '2025-01-01T00:00:00Z' }
      }
    ]
  }
]);
Copy

Sync records from a stream

Sync records from a stream using a cursor-based approach. Supports both auto-pagination and manual pagination. The sync API always returns nextCursor (for resumption) and hasNext (to indicate whether more data is immediately available). Auto-pagination uses hasNext to determine when to stop.

// Get first page with items, nextCursor, and hasNext
const response = await client.records.sync('my_stream', {
  initializeCursor: '1d-ago',
  sources: [{ source: { type: 'container', space: 'mySpace', externalId: 'myContainer' }, properties: ['*'] }],
});
console.log(response.items, response.nextCursor, response.hasNext);
Copy

Upsert records into a stream

Create or update records in a mutable stream. If a record with the same space + externalId already exists, it will be fully replaced (no partial updates).

Note: This endpoint is only available for mutable streams.

await client.records.upsert('my_mutable_stream', [
  {
    space: 'mySpace',
    externalId: 'record1',
    sources: [
      {
        source: { type: 'container', space: 'mySpace', externalId: 'myContainer' },
        properties: { temperature: 30.0, timestamp: '2025-01-01T00:00:00Z' }
      }
    ]
  }
]);
Copy