Jobs

Create sync jobs, schedule them, pause and resume, trigger on demand, and preview upcoming runs.

A job says what to pull from a source, how its fields map to signals, and when to sync. The connect flow creates jobs for you; the routes below are the direct surface for custom configs and day-two management.

Create a job

The source must exist in your workspace and be active:

curl -X POST "$API_BASE/v1/ingestion/jobs" \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Workspace-Id: $WORKSPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{
    "source_id": "3f8a1c2e-…",
    "name": "Daily sales import",
    "description": "Import sales data every night",
    "config": {
      "query": "SELECT * FROM sales WHERE date > :last_sync",
      "field_mappings": [
        { "source_field": "sale_date", "target_key": "timestamp",
          "data_type": "datetime_timestamp", "is_timestamp": true },
        { "source_field": "amount", "target_key": "revenue",
          "data_type": "continuous_numerical" }
      ],
      "options": { "delete_existing": true }
    },
    "schedule": {
      "type": "cron",
      "cron_expression": "0 2 * * *",
      "timezone": "UTC",
      "enabled": true
    }
  }'

from predictai import PredictAI

client = PredictAI(token="pa_live_…", workspace_id="ws_…")

data = client.ingestion.post_ingestion_jobs(json={
    "source_id": "3f8a1c2e-…",
    "name": "Daily sales import",
    "description": "Import sales data every night",
    "config": {
        "query": "SELECT * FROM sales WHERE date > :last_sync",
        "field_mappings": [
            {
                "source_field": "sale_date",
                "target_key": "timestamp",
                "data_type": "datetime_timestamp",
                "is_timestamp": True,
            },
            {
                "source_field": "amount",
                "target_key": "revenue",
                "data_type": "continuous_numerical",
            },
        ],
        "options": {"delete_existing": True},
    },
    "schedule": {
        "type": "cron",
        "cron_expression": "0 2 * * *",
        "timezone": "UTC",
        "enabled": True,
    },
})

import { PredictAI } from "@predictai/sdk";

const client = new PredictAI({ token: "pa_live_…", workspaceId: "ws_…" });

const data = await client.ingestion.postIngestionJobs({
  json: {
    source_id: "3f8a1c2e-…",
    name: "Daily sales import",
    description: "Import sales data every night",
    config: {
      query: "SELECT * FROM sales WHERE date > :last_sync",
      field_mappings: [
        {
          source_field: "sale_date",
          target_key: "timestamp",
          data_type: "datetime_timestamp",
          is_timestamp: true,
        },
        {
          source_field: "amount",
          target_key: "revenue",
          data_type: "continuous_numerical",
        },
      ],
      options: { delete_existing: true },
    },
    schedule: {
      type: "cron",
      cron_expression: "0 2 * * *",
      timezone: "UTC",
      enabled: true,
    },
  },
});

{
  "data": {
    "job_id": "b7d94e10-…",
    "status": "created",
    "next_run_at": "2026-07-16T02:00:00+00:00"
  }
}

Request body

Field	Meaning
`source_id` required	The source to pull from — must be `active` in this workspace.
`name` required	Display name for the job.
`config` required	What to pull and how it maps: a `query` (or `table` + `mode`), `field_mappings` (source column → signal key, with `data_type` and `is_timestamp` / `is_entity` / `is_event` flags), and `options` like `delete_existing`.
`schedule` optional	When to sync: `{ "type": "cron" \| "interval" \| "once", "cron_expression", "timezone", "enabled" }`. Omit for a manually-triggered job.
`description` optional	Free-text description.

Scheduling priority comes from your plan — a priority field in the request is ignored.

Errors

Status	Why	Example message
`404`	Source ID isn't in this workspace	`"Source not found in this workspace"`
`400`	Source is in `error` state	`"Source is not active. Test connection first."`
`400`	Cron expression didn't validate	`"Invalid cron expression: …"`

List jobs

curl "$API_BASE/v1/ingestion/jobs?source_id=$SOURCE_ID&page=1&per_page=25" \
  -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_jobs(params={"source_id": SOURCE_ID, "page": 1, "per_page": 25})

import { PredictAI } from "@predictai/sdk";

const client = new PredictAI({ token: "pa_live_…", workspaceId: "ws_…" });

const data = await client.ingestion.getIngestionJobs({ params: { source_id: SOURCE_ID, page: 1, per_page: 25 } });

{
  "data": {
    "jobs": [
      {
        "job_id": "b7d94e10-…",
        "source_id": "3f8a1c2e-…",
        "name": "Daily sales import",
        "connector_type": "postgres",
        "config": { "table": "daily_sales", "mode": "incremental", … },
        "schedule": { "type": "cron", "cron_expression": "0 2 * * *", "enabled": true },
        "status": "active",
        "next_run_at": "2026-07-16T02:00:00+00:00",
        "last_run_at": "2026-07-15T02:00:01+00:00",
        "execution_count": 41,
        "success_count": 39,
        "failure_count": 2,
        "created_at": "2026-06-30T11:20:00+00:00"
      }
    ],
    "pagination": {
      "page": 1, "per_page": 25, "total": 3, "total_pages": 1,
      "sort_column": "created_at", "sort_direction": "desc"
    }
  }
}

Query parameters

Parameter	Meaning
`source_id` optional	Only one source's jobs.
`page` / `per_page` optional	Pagination. Defaults `1` and `100`.
`sort_column` / `sort_direction` optional	Sort. Defaults `created_at`, `desc`.

Update or pause a job

Send only what changes:

Field	Meaning
`status` optional	`"paused"` stops the scheduler picking the job up; `"active"` resumes.
`schedule` optional	Replace the schedule — `next_run_at` is recalculated.
`config` optional	Replace the query / mapping configuration.
`name` / `description` optional	Rename or re-describe.

curl -X PUT "$API_BASE/v1/ingestion/jobs/$JOB_ID" \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Workspace-Id: $WORKSPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{ "status": "paused" }'

from predictai import PredictAI

client = PredictAI(token="pa_live_…", workspace_id="ws_…")

data = client.ingestion.put_ingestion_jobs_by_job_id(JOB_ID, json={"status": "paused"})

import { PredictAI } from "@predictai/sdk";

const client = new PredictAI({ token: "pa_live_…", workspaceId: "ws_…" });

const data = await client.ingestion.putIngestionJobsByJobId(JOB_ID, { json: { status: "paused" } });

{ "data": { "status": "updated" } }

Trigger a job now

Queue a run immediately, regardless of the schedule:

curl -X POST "$API_BASE/v1/ingestion/jobs/$JOB_ID/trigger" \
  -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_jobs_by_job_id_trigger(JOB_ID)

import { PredictAI } from "@predictai/sdk";

const client = new PredictAI({ token: "pa_live_…", workspaceId: "ws_…" });

const data = await client.ingestion.postIngestionJobsByJobIdTrigger(JOB_ID);

{ "data": { "status": "triggered", "message": "Job queued for execution" } }

Track the run in Executions — filter the list by this job_id.

Preview the schedule

See the next N fire times for a cron-scheduled job (n defaults to 5):

curl "$API_BASE/v1/ingestion/jobs/$JOB_ID/preview-schedule?n=3" \
  -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_jobs_by_job_id_preview_schedule(JOB_ID, params={"n": 3})

import { PredictAI } from "@predictai/sdk";

const client = new PredictAI({ token: "pa_live_…", workspaceId: "ws_…" });

const data = await client.ingestion.getIngestionJobsByJobIdPreviewSchedule(JOB_ID, { params: { n: 3 } });

{
  "data": {
    "next_runs": [
      "2026-07-16T02:00:00+00:00",
      "2026-07-17T02:00:00+00:00",
      "2026-07-18T02:00:00+00:00"
    ]
  }
}

Interval and run-once jobs don't have cron previews:

{ "data": { "message": "Job does not use cron scheduling", "schedule_type": "interval" } }

Delete a job

curl -X DELETE "$API_BASE/v1/ingestion/jobs/$JOB_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.delete_ingestion_jobs_by_job_id(JOB_ID)

import { PredictAI } from "@predictai/sdk";

const client = new PredictAI({ token: "pa_live_…", workspaceId: "ws_…" });

const data = await client.ingestion.deleteIngestionJobsByJobId(JOB_ID);

{ "data": { "status": "deleted" } }

Errors (single-job routes)

Status	Why	Example message
`404`	Job ID isn't in this workspace	`"ingestion_jobs resource not found in this workspace"`
`400`	Deleting while executions are processing	`"Cannot delete job with 1 running executions. Cancel them first."`

On this page