Ingestion
Executions
Track ingestion runs — status, progress, errors, cancellation, retries, and workspace stats.
Every run of a job (and every ad-hoc file upload) produces an
execution. Status moves queued → processing → completed,
failed, or cancelled.
List executions
curl "$API_BASE/v1/ingestion/executions?status=failed&per_page=10" \
-H "Authorization: Bearer $TOKEN" \
-H "X-Workspace-Id: $WORKSPACE_ID"from predictai import PredictAI
client = PredictAI(token="pa_live_…", workspace_id="ws_…")
data = client.ingestion.get_ingestion_executions(params={"status": "failed", "per_page": 10})
import { PredictAI } from "@predictai/sdk";
const client = new PredictAI({ token: "pa_live_…", workspaceId: "ws_…" });
const data = await client.ingestion.getIngestionExecutions({ params: { status: "failed", per_page: 10 } });
{
"data": {
"executions": [
{
"execution_id": "e5b2c9d4-…",
"job_id": "b7d94e10-…",
"source_id": "3f8a1c2e-…",
"connector_type": "postgres",
"status": "failed",
"trigger_type": "scheduled",
"started_at": "2026-07-15T02:00:01+00:00",
"completed_at": "2026-07-15T02:00:09+00:00",
"progress": { "percentage": 40 },
"stats": { "rows_processed": 0, "duration_seconds": 8.2 },
"error": {
"message": "relation \"daily_sales\" does not exist",
"phase": "fetch"
}
},
…
],
"pagination": {
"page": 1, "per_page": 10, "total": 2, "total_pages": 1,
"sort_column": "started_at", "sort_direction": "desc"
}
}
}Query parameters
| Parameter | Meaning |
|---|---|
job_id / source_id / connector_type / status optional | Filters. |
page / per_page optional | Pagination. Defaults 1 and 10. |
sort_column / sort_direction optional | Sort. Defaults started_at, desc. |
Inspect one execution
The full record — this is what a poll_url from an upload points at:
curl "$API_BASE/v1/ingestion/executions/$EXECUTION_ID" \
-H "Authorization: Bearer $TOKEN" \
-H "X-Workspace-Id: $WORKSPACE_ID"from predictai import PredictAI
client = PredictAI(token="pa_live_…", workspace_id="ws_…")
data = client.ingestion.get_ingestion_executions_by_execution_id(EXECUTION_ID)
import { PredictAI } from "@predictai/sdk";
const client = new PredictAI({ token: "pa_live_…", workspaceId: "ws_…" });
const data = await client.ingestion.getIngestionExecutionsByExecutionId(EXECUTION_ID);
{
"data": {
"execution": {
"execution_id": "e5b2c9d4-…",
"job_id": "adhoc-file-upload",
"source_id": "file-upload",
"connector_type": "file",
"status": "completed",
"trigger_type": "manual",
"started_at": "2026-07-15T10:14:02+00:00",
"completed_at": "2026-07-15T10:14:31+00:00",
"progress": { "percentage": 100 },
"stats": {
"rows_processed": 52000,
"rows_inserted": 51940,
"duration_seconds": 29.1,
"throughput_records_per_sec": 1786.9
},
"error": null
}
}
}progress tracks a running import; stats summarizes a finished one; a
failed run's error carries the message and the phase that failed
(queue, fetch, insert, …).
Cancel an execution
Only runs that haven't finished can be cancelled. Cancelling also removes the data that run had already inserted:
curl -X POST "$API_BASE/v1/ingestion/executions/$EXECUTION_ID/cancel" \
-H "Authorization: Bearer $TOKEN" \
-H "X-Workspace-Id: $WORKSPACE_ID"from predictai import PredictAI
client = PredictAI(token="pa_live_…", workspace_id="ws_…")
data = client.ingestion.post_ingestion_executions_by_execution_id_cancel(EXECUTION_ID)
import { PredictAI } from "@predictai/sdk";
const client = new PredictAI({ token: "pa_live_…", workspaceId: "ws_…" });
const data = await client.ingestion.postIngestionExecutionsByExecutionIdCancel(EXECUTION_ID);
{
"data": {
"status": "cancellation_requested",
"message": "Execution will be cancelled and all data will be removed"
}
}Errors
| Status | Why | Example message |
|---|---|---|
400 | Execution already finished | "Cannot cancel execution with status: completed" |
404 | Execution isn't in this workspace | "ingestion_executions resource not found in this workspace" |
Retry a failed execution
curl -X POST "$API_BASE/v1/ingestion/executions/$EXECUTION_ID/retry" \
-H "Authorization: Bearer $TOKEN" \
-H "X-Workspace-Id: $WORKSPACE_ID"from predictai import PredictAI
client = PredictAI(token="pa_live_…", workspace_id="ws_…")
data = client.ingestion.post_ingestion_executions_by_execution_id_retry(EXECUTION_ID)
import { PredictAI } from "@predictai/sdk";
const client = new PredictAI({ token: "pa_live_…", workspaceId: "ws_…" });
const data = await client.ingestion.postIngestionExecutionsByExecutionIdRetry(EXECUTION_ID);
Two paths, depending on where the failure came from:
- A connector job — the retry re-reads the live job and source configuration, so any edits you made since the failure are picked up:
{ "data": { "status": "retrying", "message": "New execution queued" } }- An ad-hoc file upload — the retry re-queues the same staged file from the original execution's snapshot, under a new execution ID:
{
"data": {
"status": "retrying",
"message": "Adhoc file upload re-queued",
"execution_id": "c4f01e77-…"
}
}Errors
| Status | Why | Example message |
|---|---|---|
400 | Execution isn't in failed state | "Can only retry failed executions" |
400 | The upload snapshot is gone | "Cannot retry: original execution snapshot is empty. Please re-upload the file." |
500 | The retry couldn't be queued — safe to retry | "Failed to publish retry … - please try again." |
Workspace stats
Aggregate execution counts and performance for the whole workspace:
curl "$API_BASE/v1/ingestion/executions/stats" \
-H "Authorization: Bearer $TOKEN" \
-H "X-Workspace-Id: $WORKSPACE_ID"from predictai import PredictAI
client = PredictAI(token="pa_live_…", workspace_id="ws_…")
data = client.ingestion.get_ingestion_executions_stats()
import { PredictAI } from "@predictai/sdk";
const client = new PredictAI({ token: "pa_live_…", workspaceId: "ws_…" });
const data = await client.ingestion.getIngestionExecutionsStats();
{
"data": {
"stats": {
"by_status": {
"completed": { "count": 39, "avg_duration_seconds": 12.4, "avg_throughput": 812.5 },
"failed": { "count": 2, "avg_duration_seconds": 8.2, "avg_throughput": 0 }
},
"total_executions": 41,
"total_rows_inserted": 402113
}
}
}