Connect & import

Inspect a connection, create a source with ready-to-run syncs, upload files, and stage multi-file datasets.

The connect flow is the fast path: one call to inspect, one call to create. Inspect tests the connection, introspects the schema, and returns every table already auto-mapped — timestamp, entity, types, sync frequency. Create takes your selection and produces the source, its jobs, and (optionally) an immediate historical backfill.

Inspect a connection

Paste a connection string, or send an explicit connector_type + config. Nothing is persisted:

curl -X POST "$API_BASE/v1/ingestion/connect/inspect" \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Workspace-Id: $WORKSPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{ "connection_string": "postgresql://reader:[email protected]:5432/analytics" }'

from predictai import PredictAI

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

data = client.ingestion.post_ingestion_connect_inspect(json={
    "connection_string": "postgresql://reader:[email protected]:5432/analytics",
})

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

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

const data = await client.ingestion.postIngestionConnectInspect({
  json: {
    connection_string: "postgresql://reader:[email protected]:5432/analytics",
  },
});

{
  "data": {
    "status": "success",
    "connector_type": "postgres",
    "label": "analytics on db.example.com",
    "config": { "connection": { "host": "db.example.com", "port": 5432, "database": "analytics" } },
    "test": { "success": true },
    "parse": { "connector_type": "postgres", "confidence": "high", … },
    "tables": [
      {
        "table_name": "daily_sales",
        "timestamp_column": "sale_date",
        "entity_column": "region",
        "watermark_column": "sale_date",
        "watermark_type": "datetime",
        "recommended_frequency": "1h",
        "columns": [
          { "source_field": "sale_date", "target_key": "daily_sales.sale_date",
            "data_type": "datetime_timestamp", "is_timestamp": true },
          { "source_field": "amount", "target_key": "daily_sales.amount",
            "data_type": "continuous_numerical" },
          …
        ]
      },
      …
    ],
    "table_count": 12,
    "plan": { "signals_limit": 100000, "signals_used": 41230, "signals_remaining": 58770 },
    "raw_sample": null,
    "raw_sample_truncated": false
  }
}

Request body

Field	Meaning
`connection_string` required	The connection URL to inspect. Omit it only when sending `connector_type` + `config` instead.
`connector_type` + `config` optional	Explicit configuration instead of a string.
`credentials` optional	Secrets that can't live in a URL.
`service_account_json` / `username` / `password` optional	Convenience shortcuts merged into credentials.
`connection` optional	Advanced overrides (REST auth headers, SSL mode, …).
`sample_params` optional	REST only: extra query params for the sample request — e.g. a time-window param so the sampled response isn't empty.

Databases and SaaS providers return their tables; REST and GraphQL endpoints are sampled and shape-detected instead (the exact sampled response comes back as raw_sample). A failed connection test is still a 200 — it's an expected, user-actionable outcome:

{
  "data": {
    "status": "failed",
    "connector_type": "postgres",
    "label": "analytics on db.example.com",
    "test": { "success": false, "error": "Connection refused" },
    "tables": []
  }
}

Errors

Status	Why	Example message
`400`	Connector couldn't be resolved from the input	`"Unsupported or undetected connector type: 'snowflake'"`
`400`	An OAuth `token_ref` from another workspace	`"This connected account isn't available in this workspace. Reconnect it."`
`200` (status `failed`)	Connection test failed, or a sampled REST response had nothing to map	`"The API responded, but the data array at 'bars.AAPL' is empty - …"`

Create a source with syncs

Confirm the tables (pass back the field_mappings from inspect, or omit them to re-derive), pick a frequency per table, and choose whether to backfill history:

curl -X POST "$API_BASE/v1/ingestion/connect/create" \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Workspace-Id: $WORKSPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{
    "connection_string": "postgresql://reader:[email protected]:5432/analytics",
    "name": "Analytics DB",
    "import_historical": true,
    "tables": [
      {
        "table_name": "daily_sales",
        "frequency": "1h",
        "timestamp_column": "sale_date",
        "entity_column": "region",
        "watermark_column": "sale_date",
        "watermark_type": "datetime"
      }
    ]
  }'

from predictai import PredictAI

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

data = client.ingestion.post_ingestion_connect_create(json={
    "connection_string": "postgresql://reader:[email protected]:5432/analytics",
    "name": "Analytics DB",
    "import_historical": True,
    "tables": [
        {
            "table_name": "daily_sales",
            "frequency": "1h",
            "timestamp_column": "sale_date",
            "entity_column": "region",
            "watermark_column": "sale_date",
            "watermark_type": "datetime",
        },
    ],
})

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

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

const data = await client.ingestion.postIngestionConnectCreate({
  json: {
    connection_string: "postgresql://reader:[email protected]:5432/analytics",
    name: "Analytics DB",
    import_historical: true,
    tables: [
      {
        table_name: "daily_sales",
        frequency: "1h",
        timestamp_column: "sale_date",
        entity_column: "region",
        watermark_column: "sale_date",
        watermark_type: "datetime",
      },
    ],
  },
});

{
  "data": {
    "status": "created",
    "source_id": "3f8a1c2e-…",
    "job_ids": ["b7d94e10-…"],
    "import_historical": true,
    "message": "Connected. Importing history and syncing 1 table(s)."
  }
}

Request body

Field	Meaning
`connection_string` required	Same connection input as inspect — or `connector_type` + `config` + `credentials`.
`tables` required	The tables to sync — see the per-table fields below.
`name` optional	Display name for the source.
`import_historical` optional	`true` (the default) triggers every job's backfill immediately. `false` seeds each job's watermark to "now" so the first scheduled run only pulls new rows.

Each tables[] entry:

Field	Meaning
`table_name` required	The table to sync.
`frequency` required	Sync cadence — e.g. `15m`, `1h`, `daily`. Start from `recommended_frequency` in the inspect response.
`timestamp_column` / `entity_column` / `watermark_column` / `watermark_type` optional	The mapping from inspect, passed back. Omit to re-derive.
`field_mappings` optional	Column-level mapping from inspect, with your edits. Omit to re-derive.
`backfill_limit` optional	Keep only the most recent N rows of history.
`backfill_range` optional	`{ "from": …, "to": … }` bounds on the watermark column.

Errors

Status	Why	Example message
`400`	Connector couldn't be resolved from the input	`"Unsupported or undetected connector type: 'snowflake'"`
`403`	No access to this workspace	`"Access denied to this workspace"`

Add syncs to an existing source

Re-introspect an already-connected source using its stored credentials — no need to re-send them:

curl "$API_BASE/v1/ingestion/connect/inspect-source/$SOURCE_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_connect_inspect_source_by_source_id(SOURCE_ID)

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

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

const data = await client.ingestion.getIngestionConnectInspectSourceBySourceId(SOURCE_ID);

The response has the same shape as inspect (status, test, auto-mapped tables, plan). Then attach jobs for the tables you picked:

curl -X POST "$API_BASE/v1/ingestion/connect/add-jobs/$SOURCE_ID" \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Workspace-Id: $WORKSPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{
    "import_historical": true,
    "tables": [{ "table_name": "refunds", "frequency": "daily" }]
  }'

from predictai import PredictAI

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

data = client.ingestion.post_ingestion_connect_add_jobs_by_source_id(SOURCE_ID, json={
    "import_historical": True,
    "tables": [{"table_name": "refunds", "frequency": "daily"}],
})

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

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

const data = await client.ingestion.postIngestionConnectAddJobsBySourceId(SOURCE_ID, {
  json: {
    import_historical: true,
    tables: [{ table_name: "refunds", frequency: "daily" }],
  },
});

{
  "data": {
    "status": "created",
    "source_id": "3f8a1c2e-…",
    "job_ids": ["0c31f7aa-…"],
    "message": "Added 1 sync job(s)."
  }
}

Field	Meaning
`tables` required	Same per-table shape as create.
`import_historical` optional	Backfill the new tables' history. Defaults to `true`.

Both routes return 404 "Source not found" if the ID isn't in this workspace.

Upload a file

One multipart call imports a csv, excel, json, or parquet file. Format detection and column typing happen server-side — just pick a file:

curl -X POST "$API_BASE/v1/ingestion/connect/upload" \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Workspace-Id: $WORKSPACE_ID" \
  -F "file=@daily_sales.csv"

from predictai import PredictAI

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

data = client.ingestion.post_ingestion_connect_upload(files={"file": open("daily_sales.csv", "rb")})

import { PredictAI } from "@predictai/sdk";
import { openAsBlob } from "node:fs";

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

const form = new FormData();
form.append("file", await openAsBlob("daily_sales.csv"), "daily_sales.csv");

const data = await client.ingestion.postIngestionConnectUpload({ form });

{
  "data": {
    "status": "queued",
    "execution_id": "e5b2c9d4-…",
    "filename": "daily_sales.csv",
    "file_format": "csv",
    "detected_schema": {
      "timestamp_column": "sale_date",
      "entity_column": "region",
      "columns": [
        { "name": "sale_date", "type": "datetime_timestamp" },
        { "name": "amount", "type": "continuous_numerical" },
        …
      ]
    },
    "file_size_bytes": 482113,
    "poll_url": "/v1/ingestion/executions/e5b2c9d4-…",
    "message": "File uploaded and auto-typed. Processing in background."
  }
}

The 202 means the import runs in the background — poll the poll_url (see Executions) to track progress.

Request (multipart form)

Field	Meaning
`file` required	The file to import: csv, excel, json, or parquet.
`columnTypes` optional	An explicit schema, skipping auto-detection.
`deleteExisting` optional	Delete existing values for the file's keys before importing.
`skipDuplicates` optional	Skip rows whose key + timestamp already exist.

There is also a classic endpoint, POST /v1/ingestion/sources/upload, with the same multipart contract and 202 response — but there columnTypes is required (no auto-typing).

Errors

Status	Why	Example message
`400`	No file in the request	`"No file provided"` / `"No file selected"`
`400`	The file couldn't be read	`"Could not read the file - it may be empty or corrupt."`
`400`	Classic endpoint without a schema	`"columnTypes schema required. See docs/CONNECTOR_SCHEMAS.md"`
`502`	The import couldn't be queued — safe to retry	`"Failed to publish file ingestion job …"`

Customize a file mapping

When you want to review or edit the mapping before importing, split the upload in two. First stage the file and get its detected columns:

curl -X POST "$API_BASE/v1/ingestion/connect/inspect-file" \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Workspace-Id: $WORKSPACE_ID" \
  -F "file=@daily_sales.csv"

from predictai import PredictAI

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

data = client.ingestion.post_ingestion_connect_inspect_file(files={"file": open("daily_sales.csv", "rb")})

import { PredictAI } from "@predictai/sdk";
import { openAsBlob } from "node:fs";

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

const form = new FormData();
form.append("file", await openAsBlob("daily_sales.csv"), "daily_sales.csv");

const data = await client.ingestion.postIngestionConnectInspectFile({ form });

{
  "data": {
    "status": "success",
    "file_path": "s3://…/uploads/2026/07/15/…_daily_sales.csv",
    "file_format": "csv",
    "file_size_bytes": 482113,
    "filename": "daily_sales.csv",
    "table": {
      "table_name": "file",
      "timestamp_column": "sale_date",
      "columns": [
        { "source_field": "sale_date", "target_key": "sale_date",
          "data_type": "datetime_timestamp", "is_timestamp": true, "original_index": 0 },
        …
      ]
    }
  }
}

Edit the columns as needed (rename target_key, toggle included, flag is_entity or is_event), then import with your mapping:

curl -X POST "$API_BASE/v1/ingestion/connect/import-file" \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Workspace-Id: $WORKSPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{
    "file_path": "s3://…/uploads/2026/07/15/…_daily_sales.csv",
    "file_format": "csv",
    "filename": "daily_sales.csv",
    "field_mappings": [ … ]
  }'

from predictai import PredictAI

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

data = client.ingestion.post_ingestion_connect_import_file(json={
    "file_path": "s3://…/uploads/2026/07/15/…_daily_sales.csv",
    "file_format": "csv",
    "filename": "daily_sales.csv",
    "field_mappings": […],
})

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

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

const data = await client.ingestion.postIngestionConnectImportFile({
  json: {
    file_path: "s3://…/uploads/2026/07/15/…_daily_sales.csv",
    file_format: "csv",
    filename: "daily_sales.csv",
    field_mappings: […],
  },
});

{
  "data": {
    "status": "queued",
    "execution_id": "a91d33f0-…",
    "poll_url": "/v1/ingestion/executions/a91d33f0-…",
    "message": "Importing with your custom mapping."
  }
}

Field	Meaning
`file_path` required	The staged path from `inspect-file`. Forgetting it returns `400` `"file_path required"`.
`field_mappings` required	The `columns` from inspect, with your edits — renamed `target_key`, toggled `included`, flagged `is_entity` / `is_event`.
`file_format` / `filename` optional	Echoed from the inspect response.

Multi-file datasets

Uploading files one by one can't see across them — colliding column names, duplicated quantities, reference tables. A dataset upload stages every file, then plans the whole set at once; you review the plan, then import. Four steps:

1. Stage each file

One call per file (keeps each request small):

curl -X POST "$API_BASE/v1/ingestion/connect/datasets/stage" \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Workspace-Id: $WORKSPACE_ID" \
  -F "[email protected]"

from predictai import PredictAI

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

data = client.ingestion.post_ingestion_connect_datasets_stage(files={"file": open("orders.csv", "rb")})

import { PredictAI } from "@predictai/sdk";
import { openAsBlob } from "node:fs";

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

const form = new FormData();
form.append("file", await openAsBlob("orders.csv"), "orders.csv");

const data = await client.ingestion.postIngestionConnectDatasetsStage({ form });

{
  "data": {
    "status": "staged",
    "file": {
      "filename": "orders.csv",
      "file_path": "s3://…/uploads/2026/07/15/…_orders.csv",
      "file_format": "csv",
      "file_size_bytes": 1204882,
      "row_estimate": 52000,
      "columns": [ { "name": "order_date", "universal_type": "datetime_timestamp" }, … ]
    }
  }
}

2. Plan the dataset

Send the staged descriptors back (max 25 files per dataset):

curl -X POST "$API_BASE/v1/ingestion/connect/datasets/inspect" \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Workspace-Id: $WORKSPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{ "files": [ { "filename": "orders.csv", "file_path": "s3://…", … }, … ] }'

from predictai import PredictAI

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

data = client.ingestion.post_ingestion_connect_datasets_inspect(json={
    "files": [{"filename": "orders.csv", "file_path": "s3://…", …}, …],
})

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

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

const data = await client.ingestion.postIngestionConnectDatasetsInspect({
  json: {
    files: [{ filename: "orders.csv", file_path: "s3://…", … }, …],
  },
});

{
  "data": {
    "status": "planned",
    "dataset_id": "7d20b4c6-…",
    "plan": {
      "files": [
        { "filename": "orders.csv", "file_path": "s3://…", "table": { "columns": [ … ] } },
        { "filename": "holidays.csv", "skip_file": true, "skip_reason": "reference data" }
      ],
      "warnings": []
    }
  }
}

Every decision in the plan carries a reason. You can also send the files as one multipart request instead of staging first — subject to the request body-size limit.

3. Import

Pass the plan back, with any review edits:

curl -X POST "$API_BASE/v1/ingestion/connect/datasets/import" \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Workspace-Id: $WORKSPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{ "dataset_id": "7d20b4c6-…", "plan": { … } }'

from predictai import PredictAI

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

data = client.ingestion.post_ingestion_connect_datasets_import(json={"dataset_id": "7d20b4c6-…", "plan": {…}})

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

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

const data = await client.ingestion.postIngestionConnectDatasetsImport({ json: { dataset_id: "7d20b4c6-…", plan: { … } } });

{
  "data": {
    "status": "queued",
    "dataset_id": "7d20b4c6-…",
    "executions": [
      { "filename": "orders.csv", "execution_id": "f0aa19c3-…" }
    ],
    "skipped": [
      { "filename": "holidays.csv", "reason": "skipped by plan" }
    ],
    "poll_url": "/v1/ingestion/connect/datasets/7d20b4c6-…"
  }
}

4. Poll batch status

curl "$API_BASE/v1/ingestion/connect/datasets/$DATASET_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_connect_datasets_by_dataset_id(DATASET_ID)

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

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

const data = await client.ingestion.getIngestionConnectDatasetsByDatasetId(DATASET_ID);

{
  "data": {
    "dataset_id": "7d20b4c6-…",
    "status": "importing",
    "files": [
      {
        "filename": "orders.csv",
        "execution_id": "f0aa19c3-…",
        "status": "processing",
        "progress": 62,
        "rows_inserted": 32240,
        "error": null
      }
    ],
    "skipped": [ { "filename": "holidays.csv", "reason": "skipped by plan" } ],
    "plan": { … },
    "created_at": "2026-07-15T10:12:44+00:00"
  }
}

Overall status is planned, importing, completed, completed_with_errors, or failed.

Errors (dataset routes)

Status	Why	Example message
`400`	Too many files	`"Too many files - max 25 per dataset upload."`
`400`	A descriptor wasn't produced by `datasets/stage`	`"Invalid staged file descriptor(s): orders.csv. Stage each file via /datasets/stage first."`
`400`	Nothing readable, missing `dataset_id`, or an empty plan	`"No readable files provided - stage files first or attach them."`
`404`	Dataset isn't in this workspace	`"Dataset not found"`
`409`	An import for this dataset is already running	`"Dataset import already running"`
`502`	No file could be queued	`"No files could be queued for import"`

On this page