Data sources

Create, list, inspect, update, and delete sources — plus schema introspection and connection tests.

All routes require Authorization and X-Workspace-Id headers. For the guided connect flow that creates a source and its sync jobs in one call, see Connect & import — the routes below are the direct CRUD surface.

Test a connection first

Dry-run a configuration before creating anything. Nothing is persisted:

curl -X POST "$API_BASE/v1/ingestion/sources/test-connection" \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Workspace-Id: $WORKSPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{
    "connector_type": "postgres",
    "config": {
      "connection": { "host": "db.example.com", "port": 5432, "database": "analytics" },
      "credentials": { "username": "reader", "password": "…" }
    }
  }'

from predictai import PredictAI

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

data = client.ingestion.post_ingestion_sources_test_connection(json={
    "connector_type": "postgres",
    "config": {
        "connection": {
            "host": "db.example.com",
            "port": 5432,
            "database": "analytics",
        },
        "credentials": {"username": "reader", "password": "…"},
    },
})

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

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

const data = await client.ingestion.postIngestionSourcesTestConnection({
  json: {
    connector_type: "postgres",
    config: {
      connection: {
        host: "db.example.com",
        port: 5432,
        database: "analytics",
      },
      credentials: { username: "reader", password: "…" },
    },
  },
});

{
  "data": {
    "status": "success",
    "message": "Connection test passed",
    "connector_type": "postgres",
    "details": {
      "connection_valid": true,
      "auth_valid": true,
      "permissions_valid": true
    }
  }
}

A failed test returns 400 with a categorized diagnosis so you know what to fix:

{
  "data": {
    "status": "failed",
    "error": "FATAL: password authentication failed for user \"reader\"",
    "error_details": {
      "message": "FATAL: password authentication failed for user \"reader\"",
      "connection_valid": false,
      "category": "authentication",
      "suggestion": "Check username, password, or API credentials"
    },
    "connector_type": "postgres"
  }
}

category is one of connection, authentication, authorization, timeout, ssl, not_found, or unknown.

Errors

Status	Why	Example message
`400`	Connector type isn't supported	`"Unsupported connector type: snowflake"`
`400`	The connection test failed	`"Connection refused"` (with `error_details`)
`500`	The connector itself couldn't be built	`"status": "error"` with a `system` category

Create a source

test-connection is assumed to have passed, so the source is created active immediately:

curl -X POST "$API_BASE/v1/ingestion/sources" \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Workspace-Id: $WORKSPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Analytics DB",
    "connector_type": "postgres",
    "config": {
      "connection": { "host": "db.example.com", "port": 5432, "database": "analytics" },
      "credentials": { "username": "reader", "password": "…" }
    }
  }'

from predictai import PredictAI

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

data = client.ingestion.post_ingestion_sources(json={
    "name": "Analytics DB",
    "connector_type": "postgres",
    "config": {
        "connection": {
            "host": "db.example.com",
            "port": 5432,
            "database": "analytics",
        },
        "credentials": {"username": "reader", "password": "…"},
    },
})

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

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

const data = await client.ingestion.postIngestionSources({
  json: {
    name: "Analytics DB",
    connector_type: "postgres",
    config: {
      connection: {
        host: "db.example.com",
        port: 5432,
        database: "analytics",
      },
      credentials: { username: "reader", password: "…" },
    },
  },
});

{
  "data": {
    "source_id": "3f8a1c2e-…",
    "status": "created",
    "message": "Data source created and activated successfully"
  }
}

Request body

Field	Meaning
`name` required	Display name for the source.
`connector_type` required	One of the supported connectors — see Concepts for the catalog.
`config` required	Connector configuration: `connection` (host, port, database, …) and `credentials` (username, password, API keys).

Errors

Status	Why	Example message
`400`	Connector type isn't supported	`"Unsupported connector type: snowflake"`
`400`	`X-Workspace-Id` header missing	`"Workspace ID required"`
`403`	No access to this workspace	`"Access denied to this workspace"`

List sources

curl "$API_BASE/v1/ingestion/sources" \
  -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_sources()

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

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

const data = await client.ingestion.getIngestionSources();

{
  "data": {
    "sources": [
      {
        "source_id": "3f8a1c2e-…",
        "workspace_id": "WORKSPACE_ID",
        "name": "Analytics DB",
        "connector_type": "postgres",
        "config": { "connection": { "host": "db.example.com", "port": 5432, "database": "analytics" } },
        "status": "active",
        "test_status": "success",
        "last_tested_at": "2026-07-15T08:30:00+00:00",
        "created_at": "2026-07-14T16:02:11+00:00",
        "metadata": {
          "total_jobs": 3,
          "total_executions": 41,
          "last_job_triggered": "2026-07-15T09:00:00+00:00",
          "paused_jobs": 0,
          "active_jobs": 3,
          "failed_jobs": 0,
          "pause_reason": null,
          "pause_kind": null
        }
      }
    ]
  }
}

Sources are sorted newest-first (up to 500). metadata aggregates job health at a glance — how many jobs are active, paused, or failed, and why one is paused if any are. Credentials and sensitive connection fields are never included.

Inspect one source

curl "$API_BASE/v1/ingestion/sources/$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_sources_by_source_id(SOURCE_ID)

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

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

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

{
  "data": {
    "source": { "source_id": "3f8a1c2e-…", "name": "Analytics DB", … },
    "statistics": {
      "jobs": { "total": 3, "active": 3, "scheduled": 2 },
      "executions": {
        "total": 41, "successful": 39, "failed": 2,
        "success_rate": 95.12, "last_24_hours": 6
      },
      "performance": {
        "avg_duration_seconds": 12.4,
        "avg_throughput_records_per_sec": 812.5,
        "total_rows_processed": 402113
      },
      "activity": {
        "last_execution": "2026-07-15T09:00:02+00:00",
        "last_successful_execution": "2026-07-15T09:00:14+00:00",
        "last_job_triggered": "2026-07-15T09:00:00+00:00",
        "is_active": true
      }
    }
  }
}

Errors

Status	Why	Example message
`404`	Source doesn't exist in this workspace	`"Source not found"`

Read the schema

Introspect the live structure of a connected source — databases return their tables and columns:

curl "$API_BASE/v1/ingestion/sources/$SOURCE_ID/schema" \
  -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_sources_by_source_id_schema(SOURCE_ID)

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

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

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

{
  "data": {
    "schema": {
      "tables": [
        {
          "table_name": "daily_sales",
          "columns": [
            { "name": "sale_date", "type": "timestamp" },
            { "name": "region", "type": "varchar" },
            { "name": "amount", "type": "numeric" }
          ],
          "row_estimate": 182400
        },
        …
      ]
    }
  }
}

For the same introspection plus auto-mapped field suggestions, use connect/inspect-source instead.

Re-test an existing source

Runs the connection test with the stored credentials and updates the source's test_status and status:

curl -X POST "$API_BASE/v1/ingestion/sources/$SOURCE_ID/test" \
  -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_sources_by_source_id_test(SOURCE_ID)

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

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

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

{ "data": { "status": "success", "message": "Connection test passed" } }

A failing test returns 400 with error.code bad_request and the driver message in error.message and flips the source's status to error.

Update a source

Send only the fields you're changing:

Field	Meaning
`name` optional	Rename the source.
`config` optional	Replace the connection configuration or credentials.
`status` optional	`active` or `inactive`.

curl -X PUT "$API_BASE/v1/ingestion/sources/$SOURCE_ID" \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Workspace-Id: $WORKSPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{ "name": "Analytics DB (read replica)" }'

from predictai import PredictAI

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

data = client.ingestion.put_ingestion_sources_by_source_id(SOURCE_ID, json={"name": "Analytics DB (read replica)"})

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

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

const data = await client.ingestion.putIngestionSourcesBySourceId(SOURCE_ID, { json: { name: "Analytics DB (read replica)" } });

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

Delete a source

Deleting a source also deletes its jobs. Active jobs are paused first so the scheduler can't queue new runs mid-delete:

curl -X DELETE "$API_BASE/v1/ingestion/sources/$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.delete_ingestion_sources_by_source_id(SOURCE_ID)

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

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

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

{ "data": { "status": "deleted", "paused_jobs": 3 } }

Signals already ingested are not removed — manage them in Signals.

Errors (single-source routes)

schema, test, update, and delete share one lookup rule:

Status	Why	Example message
`404`	The ID doesn't exist in this workspace (other tenants' IDs also read as not found)	`"data_sources resource not found in this workspace"`
`400`	`X-Workspace-Id` header missing	`"Workspace ID required"`

Proxy a REST source

Call a REST source's API through the platform — it injects the stored authentication, so browser clients avoid CORS and never see credentials. Works with GET or POST:

Parameter	Meaning
`endpoint` required	The path on the source's base URL to call, e.g. `/v2/metrics`.
(anything else) optional	All other query params — and any JSON body on POST — are forwarded to the upstream API unchanged.

curl "$API_BASE/v1/ingestion/sources/$SOURCE_ID/proxy?endpoint=/v2/metrics&window=7d" \
  -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_sources_by_source_id_proxy(SOURCE_ID, params={"endpoint": "/v2/metrics", "window": "7d"})

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

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

const data = await client.ingestion.getIngestionSourcesBySourceIdProxy(SOURCE_ID, { params: { endpoint: "/v2/metrics", window: "7d" } });

{
  "data": {
    "success": true,
    "data": [ { "date": "2026-07-14", "value": 1042 }, … ],
    "metadata": {
      "source_name": "Metrics API",
      "rows_fetched": 7,
      "response_time_ms": 312
    }
  }
}

Errors

Status	Why	Example message
`404`	Source doesn't exist	`"Source not found"`
`400`	Source isn't a REST connector	`"Source must be a REST connector"`
`400`	The upstream API call failed	`error.message` carrying the upstream failure

On this page