> ## Documentation Index > Fetch the complete documentation index at: https://docs.kaireonai.com/llms.txt > Use this file to discover all available pages before exploring further. # Pipelines > Create and manage ETL pipelines with visual flow nodes and edges. Supports batch and micro-batch execution modes with configurable parallelism (streaming mode is a planned placeholder). Flow Pipelines list view in the Data module **Pipeline execution modes.** `batch` and `micro_batch` are fully implemented. `streaming` is a planned placeholder — it is disabled in the UI and will not spawn a long-lived consumer at the API layer. See [Data Platform → Execution Config](/data/overview#execution-config) for details. **Coming-soon connectors.** Pipelines whose source connector is `amazon_kinesis` or `braze` will no-op at execution time. The executor logs a message and returns zero rows until ingestion support is wired. See [Connectors](/api-reference/connectors) for the full status table. ## GET /api/v1/pipelines List all pipelines with their connector and schema references. ### Query Parameters | Parameter | Type | Default | Description | | --------- | ------- | ------- | ------------------------------ | | `limit` | integer | `50` | Max results per page (max 100) | | `cursor` | string | — | Cursor for pagination | ### Response ```json theme={null} { "data": [ { "id": "pipe_001", "tenantId": "tenant_001", "name": "Customer Import", "description": "Daily customer data sync from Snowflake", "status": "draft", "schedule": "0 2 * * *", "connectorId": "conn_001", "schemaId": "schema_001", "executionConfig": { "batchSize": 5000, "parallelism": 4 }, "irVersion": "1.0", "publishedIrVersion": null, "connector": { "id": "conn_001", "name": "Production Snowflake", "type": "snowflake" }, "schema": { "id": "schema_001", "name": "customers", "displayName": "Customers" }, "lastRunAt": null, "lastRunStatus": null, "createdAt": "2026-01-15T09:00:00.000Z", "updatedAt": "2026-01-15T09:00:00.000Z" } ], "pagination": { "total": 3, "hasMore": false, "limit": 50, "cursor": null } } ``` *** ## POST /api/v1/pipelines Create a new **IR-native** pipeline. Every pipeline is IR-native — the request **must** include `irVersion: "1.0"` and an `ir` document. The legacy `nodes`/`edges` creation format was removed on 2026-04-28; a request without `irVersion: "1.0"` returns `422` with a message pointing to the IR schema. IR-native creation requires the tenant flag `flowIrEnabled`. If it is not enabled the endpoint returns `403` with `{ "error": "flow_ir_disabled" }`. Enable it via `PUT /api/v1/tenant-settings { "flowIrEnabled": true }`. ### Request Body | Field | Type | Required | Description | | ----------------- | -------------- | -------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `irVersion` | string | Yes | Must be the literal `"1.0"`. | | `ir` | object | Yes | The full Pipeline IR document (`{ version, id, metadata, nodes, ... }`). Validated with Zod + structural checks. See [Pipeline IR](/data/transforms/pipeline-ir). | | `name` | string | Yes | Pipeline name. | | `connectorId` | string | Yes | Source connector ID — must exist in the tenant. | | `schemaId` | string | Yes | Target schema ID — must exist in the tenant. | | `description` | string | No | Description. | | `schedule` | string \| null | No | Legacy cron column. Defaults to `null`; the IR's own `schedule` is the source of truth. | | `executionConfig` | object | No | Execution settings (batchSize, parallelism, partitioning). | | `authoredBy` | string \| null | No | Operator id recorded on IR version 1. | | `comment` | string | No | Optional comment recorded on IR version 1. | The validated IR is stored as **version 1** in the `pipeline_ir_versions` table. Subsequent runs via `POST /pipelines/{id}/run` are dispatched to the in-process batch interpreter. ### Example ```bash theme={null} curl -X POST https://playground.kaireonai.com/api/v1/pipelines \ -H "Content-Type: application/json" \ -H "X-Tenant-Id: my-tenant" \ -d '{ "name": "Customer Import", "connectorId": "conn_001", "schemaId": "schema_001", "schedule": "0 2 * * *", "executionConfig": { "batchSize": 5000, "parallelism": 4 }, "irVersion": "1.0", "ir": { "version": "1.0", "id": "customer-import", "metadata": { "name": "Customer Import" }, "nodes": [] } }' ``` **Response:** `201 Created` — the created pipeline object plus the stored `ir`. ### Error Responses | Status | Cause | | ------ | ---------------------------------------------------------------------------------------------------------------------------------- | | `400` | Invalid JSON, or the `ir` failed Zod/structural validation (response body includes an `errors` array). | | `403` | `flowIrEnabled` is not set for the tenant (`{ "error": "flow_ir_disabled" }`). | | `409` | A pipeline with the same `name` already exists. | | `422` | Request omits `irVersion: "1.0"`; or `name`/`connectorId`/`schemaId` missing; or `connectorId`/`schemaId` not found in the tenant. | | `429` | Rate limit exceeded (200 requests / 60s). | *** ## PUT /api/v1/pipelines Update a pipeline's metadata. IR (node/edge) mutations do **not** go through this endpoint — save a new IR version with `POST /api/v1/pipelines/{id}/ir` instead. ### Request Body | Field | Type | Required | Description | | ----------------- | -------------- | -------- | ----------------------------------------------------- | | `id` | string | Yes | Pipeline ID or name | | `name` | string | No | Updated name | | `description` | string | No | Updated description | | `schedule` | string \| null | No | Updated legacy cron column | | `executionConfig` | object | No | Updated execution config | | `status` | string | No | Updated status (`draft`, `active`, `paused`, `error`) | Every body field other than `id` is applied directly to the Pipeline row's scalar columns. Sending removed fields such as `nodes` or `edges` fails — those tables no longer exist. **Response:** `200 OK` — the updated pipeline with its `connector` and `schema` references. ### Error Responses | Status | Cause | | ------ | --------------------------------------------------------------------------------- | | `400` | Missing `id` or invalid JSON | | `404` | Pipeline not found for the caller's tenant (includes ids owned by another tenant) | The update is scoped by the `(tenantId, id)` compound key, so a `PUT` targeting a pipeline owned by another tenant returns `404` — never `200` or `500`. *** ## DELETE /api/v1/pipelines Delete a pipeline and its nodes/edges. | Parameter | Type | Required | Description | | --------- | ------ | -------- | ----------------------------- | | `id` | string | Yes | Pipeline ID (query parameter) | **Response:** `204 No Content` ### Error Responses | Status | Cause | | ------ | --------------------------------------------------------------------------------- | | `400` | Missing `id` | | `404` | Pipeline not found for the caller's tenant (includes ids owned by another tenant) | *** ## POST /api/v1/pipelines/{id}/run Run a pipeline **synchronously, in-process**. The route creates a `pipeline_run` row (status `running`), loads the published-or-latest IR version, executes it via the batch interpreter, then finalizes the run row to `completed` or `failed`. It does **not** enqueue to an external worker queue, and the response is returned only after the run finishes. ### Response (200 — success) ```json theme={null} { "runId": "8f2c1a4e-…", "irVersion": 3, "ok": true, "nodeResults": [] } ``` On failure the same shape is returned with `ok: false`, an `error` string, and `failedNodeId`, but with HTTP status `500`. ### Status Codes | Status | When | | ------ | ------------------------------------------------------------------------------------------------- | | `200` | Run completed successfully (`ok: true`) | | `404` | Pipeline not found for the tenant | | `409` | Pipeline is not IR-native (row has no `irVersion`) | | `500` | Run executed but failed (`ok: false`), or the pipeline is IR-native but has no stored IR versions | *** ## GET /api/v1/pipelines/{id}/runs List recent execution runs for a pipeline (last 20 runs, newest first). ### Response ```json theme={null} [ { "id": "run_001", "pipelineId": "pipe_001", "status": "completed", "createdAt": "2026-03-16T02:00:00.000Z", "completedAt": "2026-03-16T02:05:30.000Z" } ] ``` ## GET /api/v1/pipelines/{id}/ir/versions List the IR version history for a pipeline, descending by version number. Each save through `POST /api/v1/pipelines/{id}/ir` writes a new row to the IR version log with the next version number — this endpoint reads them back for diff/rollback UIs. The endpoint runs a tenant-scoped existence check BEFORE the version lookup so a caller from a different tenant cannot probe pipeline ids by reading an empty `[]` (200) — they get a 404 instead. ### Path Parameters | Parameter | Type | Description | | --------- | ------ | ------------------------------------------ | | `id` | string | `Pipeline.id` — must belong to the tenant. | ### Response Returned at `versions/route.ts:33-41`. The full IR is intentionally NOT included — only the version metadata. Fetch a specific IR via `GET /api/v1/pipelines/{id}/ir?version=N`. ```json theme={null} [ { "version": 7, "authoredBy": "alice@example.com", "comment": "Add filter for high-value customers", "createdAt": "2026-04-30T14:00:00.000Z" }, { "version": 6, "authoredBy": "bob@example.com", "comment": null, "createdAt": "2026-04-29T10:30:00.000Z" } ] ``` Monotonically increasing per pipeline. The first save produces `version: 1`. Operator id supplied to `savePipelineIr` at the time of the save (`pipeline-ir-repo.ts:18`). Optional human-readable comment attached at save time. ISO timestamp of when the version row was inserted. ### Status codes | Code | When | Source | | ---- | ----------------------------------------------------------------------- | ---------------------- | | 200 | Returns the version array (possibly empty for pipelines with no IR yet) | `versions/route.ts:33` | | 401 | Caller is not authenticated | `requireRole` | | 403 | Caller is not `viewer`, `editor`, or `admin` | `versions/route.ts:20` | | 404 | Pipeline not found for tenant | `versions/route.ts:30` | ### Roles admin, editor, viewer. *** ## GET /api/v1/pipelines/{id} Fetch a single pipeline's metadata (used by the flow editor topbar). Tenant-scoped; a cross-tenant id reads as `404`. ### Response `200` ```json theme={null} { "id": "pipe_001", "name": "Customer Import", "description": "Daily customer data sync from Snowflake", "status": "active", "irVersion": "1.0", "lastRunAt": "2026-03-16T02:00:00.000Z", "lastRunStatus": "completed", "connectorId": "conn_001", "schemaId": "schema_001", "createdAt": "2026-01-15T09:00:00.000Z", "updatedAt": "2026-03-16T02:05:00.000Z" } ``` Returns `404` when the pipeline does not exist for the tenant. Roles: admin, editor, viewer. *** ## GET /api/v1/pipelines/{id}/ir Return the **latest** IR document for a pipeline. ### Response `200` ```json theme={null} { "version": 3, "ir": { "version": "1.0", "id": "customer-import", "metadata": { "name": "Customer Import" }, "nodes": [] }, "createdAt": "2026-04-30T14:00:00.000Z" } ``` ### Status Codes | Status | When | | ------ | ---------------------------------------------------------- | | `200` | Returns the latest IR version | | `404` | Pipeline not found for the tenant | | `409` | Pipeline uses the legacy node/edge format (not IR-native) | | `500` | Pipeline is marked IR-native but has no stored IR versions | Roles: admin, editor, viewer. *** ## POST /api/v1/pipelines/{id}/ir Save a **new IR version** for a pipeline. The IR is validated (Zod + structural) before persistence, and a legacy pipeline is promoted to IR-native on first save. ### Request Body | Field | Type | Required | Description | | ------------ | ------ | -------- | ----------------------------------------------- | | `ir` | object | Yes | The full Pipeline IR document. | | `comment` | string | No | Human-readable comment recorded on the version. | | `authoredBy` | string | No | Operator id recorded on the version. | ### Response `201` ```json theme={null} { "version": 4, "createdAt": "2026-05-01T09:00:00.000Z" } ``` ### Status Codes | Status | When | | ------ | -------------------------------------------------------------------------- | | `201` | New IR version saved | | `400` | `ir` missing, or failed Zod/structural validation (body includes `errors`) | | `404` | Pipeline not found for the tenant | Roles: admin, editor. *** ## POST /api/v1/pipelines/{id}/publish Pin `publishedIrVersion` to a specific IR version. The scheduler and the run route both prefer the pinned version, so authors can save draft IR versions that don't go live until they publish. ### Request Body | Field | Type | Required | Description | | --------- | ------ | -------- | --------------------------------------------------------- | | `version` | number | No | Version to publish. Defaults to the latest saved version. | ### Response `200` ```json theme={null} { "publishedIrVersion": 4, "latestVersion": 4, "isLatest": true } ``` ### Status Codes | Status | When | | ------ | ----------------------------------------------------------------------- | | `200` | Version pinned | | `400` | Pipeline has no IR versions to publish, or `version` is invalid (`< 1`) | | `404` | Pipeline not found, or the requested version does not exist | Roles: admin, editor. *** ## POST /api/v1/pipelines/{id}/ir/versions/{version}/restore Copy a prior IR version forward as a **new** version (the audit trail stays intact — the source version is not mutated). The new version's comment records the source. ### Path Parameters | Parameter | Type | Description | | --------- | ------- | ------------------------------ | | `id` | string | Pipeline ID | | `version` | integer | Source version to restore from | ### Response `201` ```json theme={null} { "sourceVersion": 2, "newVersion": 5, "ir": { "version": "1.0", "id": "customer-import", "metadata": { "name": "Customer Import" }, "nodes": [] } } ``` ### Status Codes | Status | When | | ------ | -------------------------------------------------------- | | `201` | Source version re-appended as a new version | | `400` | `version` is not a finite integer `>= 1` | | `404` | Pipeline not found, or the source version does not exist | | `422` | The stored IR for that version no longer parses | Roles: admin, editor. *** ## GET /api/v1/pipelines/{id}/sql-preview For each `target` node in the pipeline's latest IR, compute the exact SQL the runtime would execute (`INSERT ... SELECT`, optionally with `TRUNCATE` or `ON CONFLICT`). Reuses the runtime's own SQL builders so the preview matches behavior 1:1. ### Response `200` ```json theme={null} { "pipelineId": "pipe_001", "version": 3, "sourceTable": "_flow_src_preview", "targets": [ { "nodeId": "load_customers", "loadMode": "append", "targetSchema": "public", "targetTable": "customers", "sql": ["INSERT INTO \"public\".\"customers\" (...) SELECT ... FROM \"_flow_src_preview\""], "warnings": [] } ] } ``` `blue_green`, `incremental_watermark`, and `cdc_mirror` targets return `sql: null` with an `unsupported` note — those modes render only at runtime via their dedicated load-mode helpers, so no static preview is fabricated. Returns `404` when the pipeline is not found or has no IR yet. Roles: admin, editor, viewer. *** ## Roles | Endpoint | Allowed Roles | | ---------------------------------------------------- | --------------------- | | `GET /pipelines` | admin, editor, viewer | | `POST /pipelines` | admin, editor | | `PUT /pipelines` | admin, editor | | `DELETE /pipelines` | admin, editor | | `GET /pipelines/{id}` | admin, editor, viewer | | `POST /pipelines/{id}/run` | admin, editor | | `GET /pipelines/{id}/runs` | any authenticated | | `GET /pipelines/{id}/ir` | admin, editor, viewer | | `POST /pipelines/{id}/ir` | admin, editor | | `GET /pipelines/{id}/ir/versions` | admin, editor, viewer | | `POST /pipelines/{id}/ir/versions/{version}/restore` | admin, editor | | `POST /pipelines/{id}/publish` | admin, editor | | `GET /pipelines/{id}/sql-preview` | admin, editor, viewer | See also: [Data Platform](/data/overview)