Pipelines & Executions API

Manage pipelines, trigger runs and fetch rows.

A pipeline definition is the data source plus an ordered list of transformation steps stored as JSON. The list order is the execution order.

List pipelines

GET/api/pipeline/pipelines/Auth required

Each record includes a derived status and last_run taken from the most recent execution, plus has_dataset, which is true once any execution has succeeded. A materialised dataset outlives the run that wrote it, so a later failure does not clear has_dataset - the analytics picker and the dataset rows endpoint both key off it rather than off status.

Create a pipeline

POST/api/pipeline/pipelines/Auth required

Creates a pipeline bound to one data source.

Request

curl -X POST https://api.introzen.com/api/pipeline/pipelines/ \
  -H "Content-Type: application/json" \
  -H "X-Tenant-ID: 42" \
  -b cookies.txt \
  -d '{
    "name": "Monthly sales",
    "description": "Clean and sort the monthly export",
    "data_source": 7,
    "is_enabled": true,
    "frequency": "daily",
    "run_time": "02:00",
    "transformations": [
      { "transformation_type": "convert_type", "column": "amount", "converted_to": "decimal" },
      { "transformation_type": "filter", "column": "amount", "operator_type": ">", "value": "100" },
      { "transformation_type": "rename", "from": "cust_id", "to": "customer_id" },
      { "transformation_type": "sort", "column": "amount", "order": "desc" }
    ]
  }'

Transformation step fields

Field	Type	Description
transformation_type	enum	filter, rename, convert_type or sort.
column	string	Target column for filter, convert_type and sort.
operator_type	enum	Comparison for filter: >, <, ==, >=, <=, !=, contains, startswith, endswith.
value	string	Comparison value for filter.
from / to	string	Source and target column names for rename.
converted_to	enum	Target type for convert_type: string, number, decimal, boolean, date.
order	enum	asc or desc for sort.

Update a pipeline

PATCH/api/pipeline/pipelines/{id}/Auth required

Partial update. Replacing transformations replaces the whole step list; historical executions keep the steps they ran with.

Execute

POST/api/pipeline/pipelines/{id}/execute/Auth required

Queues a run. Returns the new execution identifier.

202 Accepted

{ "execution_id": 431, "pipeline_id": 12, "status": "queued" }

Executions

GET/api/pipeline/executions/?pipeline={id}Auth required

Lists executions, newest first, with status, timings, record count, success rate and log.

POST/api/pipeline/executions/{id}/retry/Auth required

Starts a fresh run based on a previous execution.

POST/api/pipeline/executions/{id}/cancel/Auth required

Terminates a running execution and marks it as cancelled.

Live telemetry socket

GET/ws/pipeline/{id}/?tenant={workspaceId}Auth required

Upgrades to a WebSocket that pushes this pipeline's status and log lines as the worker emits them. The socket is a convenience, not the record: every message it carries is also readable from the executions endpoint, which is what the client falls back to when a socket cannot be established.

Authentication

Field	Type	Description
access cookie	string	Same HttpOnly JWT cookie the REST API uses. A browser sends it automatically; the socket is anonymous without it.
tenant	string	Optional query parameter selecting the active workspace. It can only narrow access - the pipeline must belong to it. Browsers cannot set a header on a WebSocket handshake, so this travels in the query string.

Messages are JSON. The server sends one transport frame when the subscription is established, then a log frame per update; a client may send {"type":"ping"} and receives {"type":"pong"}.

Frames

{ "type": "transport", "state": "live", "pipeline_id": 12, "pipeline_name": "Daily revenue" }
{ "type": "log", "log": { "status": "running", "progress": 40, "message": "Extracting rows" } }
{ "type": "log", "log": { "status": "success", "progress": 100, "message": "Execution completed" } }

A rejection is always delivered as an application close code, after the socket is accepted, so the browser can read it rather than reporting an opaque 1006:

Close codes

Field	Type	Description
4401	string	Session expired. Sign in again.
4403	string	The pipeline is in a workspace the caller cannot access.
4404	string	The pipeline does not exist.
4500	string	The channel layer is unreachable. Retry later, or poll the executions endpoint - the run is unaffected.

Read dataset rows

GET/api/pipeline/pipelines/{id}/rows/?start=0&limit=100Auth required

Streams rows from the Parquet dataset of the latest successful execution.

Query parameters

Field	Type	Description
start	integer	Zero-based offset. Defaults to 0.
limit	integer	Rows to return, 1 to 5000. Defaults to 100.

200 OK

{
  "execution_id": 428,
  "start": 0,
  "limit": 100,
  "returned": 100,
  "rows": [{ "order_date": "2026-09-01", "amount": 250.0 }]
}

Last reviewed October 2026.