Records API Reference¶

Records are the actual data entries in Centrali. Each record belongs to a Collection (schema) and contains the data fields defined by that collection.

Core Features¶

• CRUD Operations - Create, read, update, and delete records
• Upsert - Atomic create-or-update with advisory locking
• Bulk Operations - Process multiple records in a single request
• Version History - Track all changes with automatic versioning
• Change Logs - Detailed audit trail of who changed what and when
• Validation - Automatic validation against collection rules
• Relationships - Reference other records and collections
• Querying - Canonical records query surface for filtering, text search, sorting, pagination, and projections
• Real-time Events - Webhooks and triggers on data changes

Authentication¶

All Records API endpoints require authentication:

Authorization: Bearer YOUR_JWT_TOKEN

Use a JWT from a service account or an external auth provider. If you need a refresher on workspace slugs, record IDs, and recordSlug values, see the Identifier Reference.

Base URL¶

https://api.centrali.io/data/workspace/{workspace}/api/v1

Replace {workspace} with your workspace slug.

Create a Record¶

Single Record Creation¶

POST /records

Creates a new record based on a structure.

Request Body¶

{
  "structureId": "str_abc123",
  "data": {
    "field1": "value1",
    "field2": "value2"
  },
  "metadata": {
    "source": "api",
    "userId": "user123"
  }
}

TTL Query Parameters¶

Set expiration on the created record using query parameters:

If neither is provided and the structure has a defaultTtlSeconds, the record inherits that TTL.

Example with TTL:

curl -X POST "https://api.centrali.io/data/workspace/acme/api/v1/records?ttlSeconds=3600" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "structureId": "str_sessions",
    "data": {
      "userId": "user-123",
      "token": "abc-xyz"
    }
  }'

Response¶

{
  "id": "rec_xyz789",
  "structureId": "str_abc123",
  "workspaceId": "ws_123",
  "data": {
    "field1": "value1",
    "field2": "value2"
  },
  "metadata": {
    "source": "api",
    "userId": "user123"
  },
  "version": 1,
  "createdAt": "2024-01-15T10:30:00Z",
  "updatedAt": "2024-01-15T10:30:00Z",
  "createdBy": "user_abc",
  "updatedBy": "user_abc",
  "expiresAt": null
}

Example¶

curl -X POST "https://api.centrali.io/data/workspace/acme/api/v1/records" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "structureId": "str_product",
    "data": {
      "name": "Premium Widget",
      "price": 99.99,
      "inStock": true,
      "category": "electronics"
    }
  }'

Bulk Record Creation¶

POST /records/bulk

Create multiple records in a single request.

Request Body¶

{
  "structureId": "str_abc123",
  "records": [
    {
      "data": {
        "name": "Record 1"
      }
    },
    {
      "data": {
        "name": "Record 2"
      }
    }
  ],
  "options": {
    "skipValidation": false,
    "returnRecords": true
  }
}

Response¶

{
  "success": true,
  "created": 2,
  "failed": 0,
  "records": [...],
  "errors": []
}

Read Records¶

Get a Single Record¶

GET /records/{recordId}

Retrieve a specific record by ID.

Path Parameters¶

Query Parameters¶

Response¶

{
  "id": "rec_xyz789",
  "structureId": "str_abc123",
  "data": {
    "field1": "value1",
    "field2": "value2"
  },
  "version": 2,
  "createdAt": "2024-01-15T10:30:00Z",
  "updatedAt": "2024-01-15T11:30:00Z"
}

Example¶

# Get latest version
curl "https://api.centrali.io/data/workspace/acme/api/v1/records/rec_xyz789" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN"

# Get specific version
curl "https://api.centrali.io/data/workspace/acme/api/v1/records/rec_xyz789?version=1" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN"

# Include history
curl "https://api.centrali.io/data/workspace/acme/api/v1/records/rec_xyz789?includeHistory=true" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN"

List Records¶

GET /records/slug/{recordSlug}

List records from a collection slug with filtering, pagination, sorting, search, and field expansion.

Query Parameters¶

Response¶

{
  "data": [
    {
      "id": "rec_123",
      "recordSlug": "products",
      "data": {
        "name": "Widget"
      }
    }
  ],
  "meta": {
    "limit": 25,
    "offset": 0,
    "hasMore": false,
    "total": 1
  }
}

Example¶

# Get all products
curl "https://api.centrali.io/data/workspace/acme/api/v1/records/slug/products" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN"

# Filter and sort
curl "https://api.centrali.io/data/workspace/acme/api/v1/records/slug/products?data.inStock=true&data.price[gt]=50&sort=-data.price&pageSize=10" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN"

# Search
curl "https://api.centrali.io/data/workspace/acme/api/v1/records/slug/products?search=widget&searchFields=data.name,data.description" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN"

Upsert a Record¶

POST /records/slug/{recordSlug}/upsert

Atomically find a record by match fields and update it, or create a new one if no match exists. Uses PostgreSQL advisory locking to prevent race conditions, making it safe for concurrent calls with the same match criteria.

The caller needs both CREATE and UPDATE permissions, since the outcome depends on whether a matching record already exists.

Request Body¶

{
  "match": {
    "externalId": "ext-123"
  },
  "data": {
    "externalId": "ext-123",
    "name": "Widget",
    "price": 9.99
  }
}

Response¶

HTTP 201 — Record was created:

{
  "data": {
    "id": "rec_abc123",
    "data": {
      "externalId": "ext-123",
      "name": "Widget",
      "price": 9.99
    },
    "version": 1,
    "createdAt": "2026-02-12T10:30:00Z"
  },
  "operation": "created"
}

HTTP 200 — Existing record was updated:

{
  "data": {
    "id": "rec_abc123",
    "data": {
      "externalId": "ext-123",
      "name": "Widget",
      "price": 12.99
    },
    "version": 2,
    "updatedAt": "2026-02-12T11:00:00Z"
  },
  "operation": "updated"
}

Example¶

curl -X POST "https://api.centrali.io/data/workspace/acme/api/v1/records/slug/products/upsert" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "match": { "sku": "WDG-001" },
    "data": {
      "sku": "WDG-001",
      "name": "Premium Widget",
      "price": 29.99,
      "inStock": true
    }
  }'

Idempotency¶

You can pass an Idempotency-Key header for safe HTTP retries. If the same key is sent again within the deduplication window, the original response is returned without re-executing the operation.

curl -X POST "https://api.centrali.io/data/workspace/acme/api/v1/records/slug/products/upsert" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: import-batch-42-sku-WDG-001" \
  -d '{
    "match": { "sku": "WDG-001" },
    "data": { "sku": "WDG-001", "name": "Premium Widget", "price": 29.99 }
  }'

Use Cases¶

• External data sync — Sync records from webhooks, scheduled imports, or API polling using a stable external ID as the match key
• Event aggregation — Accumulate counters, rollups, or running totals in compute functions
• Idempotent writes — Safely retry writes in compute functions with at-least-once delivery without creating duplicates
• Deduplication — Ensure exactly one record exists per business key

Behavior Notes¶

• Atomicity: The entire operation runs inside a database transaction with a PostgreSQL advisory lock keyed on the match criteria. Concurrent calls with the same match fields are serialized automatically.
• Match field protection: On update, match field values are stripped from the data payload so the business key cannot be accidentally overwritten.
• Type coercion: JSONB comparison means { "count": 42 } and { "count": "42" } are treated as different values.
• Null matching: A match value of null matches records where the field exists with a JSON null value.
• Versioning: Both create and update paths increment the record version and create changelog entries.

Update Records¶

Update a Single Record¶

PATCH /records/{recordId}

Update specific fields of a record.

Request Body¶

{
  "data": {
    "field1": "new value",
    "field2": null
  },
  "metadata": {
    "reason": "Price update"
  },
  "options": {
    "createVersion": true,
    "skipValidation": false
  }
}

TTL Query Parameters¶

Modify expiration on the updated record using query parameters:

Example — extend session by 2 hours:

curl -X PATCH "https://api.centrali.io/data/workspace/acme/api/v1/records/rec_xyz789?ttlSeconds=7200" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{ "data": {} }'

Example — remove TTL:

curl -X PATCH "https://api.centrali.io/data/workspace/acme/api/v1/records/rec_xyz789?clearTtl=true" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{ "data": {} }'

Response¶

{
  "id": "rec_xyz789",
  "structureId": "str_abc123",
  "data": {
    "field1": "new value",
    "field2": null,
    "field3": "unchanged"
  },
  "version": 2,
  "previousVersion": 1,
  "updatedAt": "2024-01-15T12:00:00Z",
  "updatedBy": "user_abc"
}

Example¶

curl -X PATCH "https://api.centrali.io/data/workspace/acme/api/v1/records/rec_xyz789" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "data": {
      "price": 89.99,
      "inStock": false
    },
    "metadata": {
      "reason": "Sale pricing"
    }
  }'

Replace a Record¶

PUT /records/{recordId}

Replace all fields of a record.

Request Body¶

{
  "data": {
    "field1": "value1",
    "field2": "value2"
  }
}

All fields not specified will be removed.

Bulk Update¶

PATCH /records/bulk

Update multiple records at once.

Request Body¶

{
  "updates": [
    {
      "id": "rec_123",
      "data": {
        "status": "archived"
      }
    },
    {
      "id": "rec_456",
      "data": {
        "status": "archived"
      }
    }
  ],
  "options": {
    "skipValidation": false,
    "createVersions": true
  }
}

Response¶

{
  "success": true,
  "updated": 2,
  "failed": 0,
  "results": [...],
  "errors": []
}

Delete Records¶

Delete a Single Record¶

DELETE /records/{recordId}

Delete a record (soft delete by default).

Query Parameters¶

Response¶

{
  "success": true,
  "id": "rec_xyz789",
  "deletedAt": "2024-01-15T13:00:00Z"
}

Example¶

# Soft delete
curl -X DELETE "https://api.centrali.io/data/workspace/acme/api/v1/records/rec_xyz789" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN"

# Permanent delete
curl -X DELETE "https://api.centrali.io/data/workspace/acme/api/v1/records/rec_xyz789?permanent=true" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN"

Bulk Delete¶

DELETE /records/bulk

Delete multiple records.

Request Body¶

{
  "ids": ["rec_123", "rec_456", "rec_789"],
  "permanent": false
}

Version History¶

Get Record History¶

GET /records/{recordId}/history

Get the complete version history of a record.

Response¶

{
  "versions": [
    {
      "version": 3,
      "data": {...},
      "updatedAt": "2024-01-15T13:00:00Z",
      "updatedBy": "user_abc",
      "changes": {
        "price": {
          "old": 99.99,
          "new": 89.99
        }
      }
    },
    {
      "version": 2,
      "data": {...},
      "updatedAt": "2024-01-15T12:00:00Z"
    },
    {
      "version": 1,
      "data": {...},
      "createdAt": "2024-01-15T10:00:00Z"
    }
  ]
}

Restore Version¶

POST /records/{recordId}/restore

Restore a record to a previous version.

Request Body¶

{
  "version": 2,
  "reason": "Reverting price change"
}

Response¶

{
  "id": "rec_xyz789",
  "version": 4,
  "restoredFrom": 2,
  "data": {...}
}

Change Logs¶

Get Record Changelog¶

GET /records/{recordId}/changelog

Get detailed change history with who, what, when.

Query Parameters¶

Response¶

{
  "entries": [
    {
      "id": "log_abc123",
      "recordId": "rec_xyz789",
      "action": "update",
      "userId": "user_abc",
      "userName": "John Doe",
      "timestamp": "2024-01-15T13:00:00Z",
      "changes": {
        "price": {
          "field": "price",
          "oldValue": 99.99,
          "newValue": 89.99
        }
      },
      "metadata": {
        "reason": "Sale pricing",
        "ipAddress": "192.168.1.1"
      }
    }
  ]
}

Advanced Queries¶

Canonical Query Endpoint¶

POST /records/query

Execute a canonical query body against a collection. Use this endpoint when you need nested boolean logic, a text clause, projections, or includes.

page.withTotal is opt-in on this route. When omitted, the response still includes meta.hasMore but skips meta.total.

Request Body¶

{
  "resource": "products",
  "where": {
    "and": [
      { "data.inStock": { "eq": true } },
      { "data.price": { "gt": 50 } }
    ]
  },
  "text": {
    "query": "wireless"
  },
  "sort": [
    { "field": "createdAt", "direction": "desc" }
  ],
  "page": {
    "limit": 10,
    "withTotal": true
  },
  "select": {
    "fields": ["id", "data.name", "data.price", "createdAt"]
  },
  "include": [
    { "relation": "categoryId" }
  ]
}

Include Clause¶

Each include entry supports:

• relation — reference property on the parent collection
• where — optional filter evaluated against the target collection's namespace
• select.fields — optional projection applied to the attached related row
• include — optional nested includes

Example:

{
  "include": [
    {
      "relation": "categoryId",
      "where": { "data.active": { "eq": true } },
      "select": { "fields": ["id", "data.name", "data.ownerId"] },
      "include": [
        {
          "relation": "ownerId",
          "select": { "fields": ["id", "data.email"] }
        }
      ]
    }
  ]
}

Rules enforced by the backend:

• where filters fetched related rows only; parent rows still return.
• Singular relations that do not match attach as null; many-to-many relations attach as [].
• If the top-level query uses select.fields, each include path must be covered by that projection.
• If an include has its own select.fields, any nested include path must be covered by that local projection.
• Default nested include depth is 3. Operators can override it with DATA_QUERY_INCLUDE_MAX_DEPTH.
• Cycle detection and include-depth overflow return UNSUPPORTED_CLAUSE; missing relations or uncovered projection paths return INVALID_QUERY.

Operators¶

• Comparison: eq, ne, gt, gte, lt, lte
• List: in, nin
• String: contains, startsWith, endsWith
• Array: hasAny, hasAll
• Existence: exists
• Boolean tree: and, or, not

Query Guardrails¶

• page.withTotal defaults to false on POST /records/query
• When withTotal is omitted, the backend skips the extra count(*) query and derives hasMore with a limit + 1 probe
• Ad-hoc record queries run under a server-side statement timeout. By default this limit is 5000 ms
• Timeouts return HTTP 504 with error code GATEWAY_TIMEOUT

Response¶

{
  "data": [
    {
      "id": "rec_123",
      "data": {
        "name": "Wireless Widget",
        "price": 99.99
      }
    }
  ],
  "meta": {
    "limit": 10,
    "offset": 0,
    "hasMore": false,
    "total": 1,
    "processingTimeMs": 6,
    "mode": "hybrid"
  }
}

If page.withTotal is omitted, meta.total is not returned.

Query Test Endpoint¶

POST /records/query/test

Validate and plan a canonical query without executing it. This is useful for builders, preview flows, and query authoring tools.

Examples¶

curl -X POST "https://api.centrali.io/data/workspace/acme/api/v1/records/query/test" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "resource": "orders",
    "where": {
      "data.status": { "eq": "paid" }
    },
    "page": {
      "limit": 5
    }
  }'

Relationships¶

Reference Fields¶

Create relationships between records using reference fields:

{
  "structureId": "str_order",
  "data": {
    "customerId": "rec_customer123",  // Reference to Customer record
    "items": ["rec_item1", "rec_item2"],  // Array of references
    "shippingAddress": {  // Embedded reference
      "addressId": "rec_address456"
    }
  }
}

Expanding References¶

GET /records/{recordId}?expand=customerId,items

Automatically expand referenced records in the response:

{
  "id": "rec_order789",
  "data": {
    "customerId": {
      "id": "rec_customer123",
      "data": {
        "name": "John Doe",
        "email": "john@example.com"
      }
    },
    "items": [
      {
        "id": "rec_item1",
        "data": {
          "name": "Widget",
          "price": 29.99
        }
      }
    ]
  }
}

Webhooks & Events¶

Records API automatically triggers events that can be captured by webhooks or compute functions:

Events¶

• record.created - New record created (also emitted on upsert-create)
• record.updated - Record updated (also emitted on upsert-update)
• record.deleted - Record deleted
• record.restored - Record restored from version
• record.bulk.created - Bulk create completed
• record.bulk.updated - Bulk update completed
• record.bulk.deleted - Bulk delete completed
• record.expired - Record automatically deleted by TTL expiration

Event Payload¶

{
  "event": "record.updated",
  "timestamp": "2024-01-15T13:00:00Z",
  "workspace": "acme",
  "data": {
    "record": {
      "id": "rec_xyz789",
      "structureId": "str_product",
      "data": {...}
    },
    "previous": {
      "data": {...}
    },
    "changes": {
      "price": {
        "old": 99.99,
        "new": 89.99
      }
    },
    "metadata": {
      "userId": "user_abc",
      "source": "api"
    }
  }
}

Error Handling¶

Error Response Format¶

{
  "error": {
    "code": "VALIDATION_ERROR",
    "message": "Record validation failed",
    "details": {
      "field": "price",
      "constraint": "minimum",
      "value": -10,
      "message": "Price must be greater than 0"
    }
  }
}

Common Error Codes¶

Best Practices¶

Pagination¶

Always use pagination for large datasets:

async function getAllRecords(structureId) {
  let allRecords = [];
  let cursor = null;

  do {
    const response = await fetch(
      `/records?structureId=${structureId}&limit=100${cursor ? `&cursor=${cursor}` : ''}`,
      { headers: { 'Authorization': `Bearer ${API_KEY}` } }
    );

    const data = await response.json();
    allRecords = allRecords.concat(data.results);
    cursor = data.pagination.nextCursor;
  } while (cursor);

  return allRecords;
}

Batch Operations¶

Use bulk endpoints for multiple operations:

// Instead of this
for (const record of records) {
  await createRecord(record);
}

// Do this
await createRecordsBulk(records);

Version Control¶

Track important changes with versioning:

// Enable versioning for critical updates
await updateRecord(recordId, {
  data: { status: 'approved' },
  options: { createVersion: true },
  metadata: { reason: 'Manager approval', approvedBy: userId }
});

Error Recovery¶

Implement retry logic for transient failures:

async function createRecordWithRetry(data, maxRetries = 3) {
  for (let i = 0; i < maxRetries; i++) {
    try {
      return await createRecord(data);
    } catch (error) {
      if (error.code === 'RATE_LIMIT' && i < maxRetries - 1) {
        await sleep(Math.pow(2, i) * 1000); // Exponential backoff
      } else {
        throw error;
      }
    }
  }
}

Rate Limits¶

Exceeding limits returns HTTP 429 with retry-after header.

SDK Examples¶

JavaScript/TypeScript¶

import { CentraliSDK } from '@centrali-io/centrali-sdk';

const client = new CentraliSDK({
  baseUrl: 'https://centrali.io',
  workspaceId: 'acme',
  clientId: process.env.CENTRALI_CLIENT_ID,
  clientSecret: process.env.CENTRALI_CLIENT_SECRET
});

// Create
const record = await client.createRecord('products', {
  name: 'Widget',
  price: 99.99
});

// Read
const product = await client.getRecord('products', 'rec_xyz789');

// Update
await client.updateRecord('products', 'rec_xyz789', {
  price: 89.99
});

// Upsert
const upsertResult = await client.upsertRecord('products', {
  match: { sku: 'WDG-001' },
  data: { sku: 'WDG-001', name: 'Widget', price: 29.99 }
});
// upsertResult.operation → 'created' or 'updated'

// Query
const results = await client.records.query('products', {
  where: {
    'data.price': { lt: 100 }
  },
  sort: [{ field: 'createdAt', direction: 'desc' }],
  page: { limit: 25 }
});

// Delete
await client.deleteRecord('products', 'rec_xyz789');

Python¶

import os
import requests

base_url = "https://api.centrali.io"
workspace = "acme"
token = os.environ["CENTRALI_ACCESS_TOKEN"]

response = requests.get(
    f"{base_url}/data/workspace/{workspace}/api/v1/records/slug/products",
    headers={"Authorization": f"Bearer {token}"},
    params={"data.price[lt]": 100, "sort": "-createdAt"},
)

results = response.json()
print(results)

Related Documentation¶

• Collections API - Define your data schemas
• Query Records - Canonical querying, filters, and text search
• Functions - Process record events
• Webhooks - Real-time notifications
• Access Control - Permissions and security
• Record TTL - Automatic record expiration