> ## Documentation Index
> Fetch the complete documentation index at: https://docs.popsink.com/llms.txt
> Use this file to discover all available pages before exploring further.

# API tutorial

> Drive Popsink programmatically: authenticate, then create connectors, subscribe their data models to targets and operate the workers over the REST API.

This guide walks through the full lifecycle of a replication over the Popsink REST API: authenticate, pick an environment and a team, create a source and a target connector, subscribe the source's data models to the target, then start and monitor the workers.

<Note>
  Every Popsink instance serves its own live OpenAPI schema. It is the source of truth for request and response shapes:

  * Swagger UI: `https://<your-instance>/api/docs`
  * OpenAPI spec: `https://<your-instance>/api/openapi.json`
</Note>

## Quick start

```bash theme={null}
export POPSINK_URL="https://<tenant>.<region>.popsink.com"

# 1. Get a token (email + password)
TOKEN=$(curl -sS -X POST "$POPSINK_URL/api/auth/jwt/login" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  --data-urlencode "username=user@example.com" \
  --data-urlencode "password=secure123" | jq -r .access_token)

# 2. Use it on every call
curl -sS "$POPSINK_URL/api/envs/" -H "Authorization: Bearer $TOKEN"
```

***

## Concepts

```
environment
  └── team
        ├── source connector          reads a source system (CDC, batch or Kafka)
        │     └── data model          one per replicated table / collection / topic
        │           └── subscription  delivery of that data model to one target
        └── target connector          writes to the destination
```

| Concept | Description |
| - | - |
| **Environment** | Top-level scope. Nearly every list endpoint is filtered by environment |
| **Team** | A group of users inside an environment. Connectors belong to a team |
| **Connector** | A source or a target, identified by its `connector_type` (e.g. `POSTGRES_SOURCE`, `SNOWFLAKE_TARGET`) |
| **Data model** | One stream produced by a source connector: a table, a collection or a Kafka topic. Created automatically from the source's table list |
| **Subscription** | Delivers one data model to one target connector, with its target table, column mapping, backfill and error-table settings |

***

## API overview

### Base URL

All endpoints live under `/api`:

```
https://<tenant>.<region>.popsink.com/api/
```

Trailing slashes are part of the path: use `/connectors/`, not `/connectors`.

### Authentication

Every endpoint except login requires a bearer token:

```
Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...
```

### Pagination

List endpoints are paginated with `page` (starting at `1`) and `size` (default `50`, max `100`). They return:

```json theme={null}
{
  "items": [ ... ],
  "total": 132,
  "page": 1,
  "size": 50,
  "pages": 3
}
```

### Idempotent creation

`POST /envs/`, `POST /teams/` and `POST /connectors/` accept an `Idempotency-Key` header. A retried call with the same key and body returns the original response instead of creating a duplicate:

```bash theme={null}
-H "Idempotency-Key: 7f9c2ba4-e88f-11ee-a951-0242ac120002"
```

### HTTP status codes

| Code | Meaning | When you'll see it |
| - | - | - |
| `200` | Success | `GET`, `PATCH` |
| `201` | Created | `POST` that creates a resource |
| `202` | Accepted | Asynchronous operations (start/stop a connector, a data model or a subscription) |
| `204` | No Content | `DELETE`, bulk member creation |
| `400` | Bad Request | Wrong password, invalid input |
| `401` | Unauthorized | Missing or expired token |
| `403` | Forbidden | You are not a member of the team that owns the resource |
| `404` | Not Found | Unknown resource |
| `409` | Conflict | A resource with the same name already exists |
| `422` | Validation Error | The body does not match the schema |
| `429` | Too Many Requests | Too many failed login attempts |

***

## Step 1: Authenticate

You can get a token in two ways.

### Option A: email and password

```bash theme={null}
curl -X POST "$POPSINK_URL/api/auth/jwt/login" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  --data-urlencode "username=john.doe@company.com" \
  --data-urlencode "password=MySecurePassword123!"
```

**Response (200 OK):**

```json theme={null}
{
  "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
  "refresh_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
  "token_type": "bearer"
}
```

When the access token expires, exchange the refresh token for a new pair:

```bash theme={null}
curl -X POST "$POPSINK_URL/api/auth/jwt/refresh" \
  -H "Content-Type: application/json" \
  -d '{"refresh_token": "'"$REFRESH_TOKEN"'"}'
```

### Option B: organization API key (self-hosted)

Create an API key in the control plane under **Dashboard → API keys → Create**. It starts with `psk_`, is shown once, and grants access to a chosen set of environments.

```bash theme={null}
curl -X POST "$POPSINK_URL/api/auth/jwt/login-from-api-key" \
  -H "Content-Type: application/json" \
  -d '{"api_key": "'"$POPSINK_API_KEY"'"}'
```

The returned token lasts **10 minutes** and comes with no refresh token: exchange the key again when you get a `401`.

<Warning>
  API-key login only works on deployments running in self-hosted mode. Otherwise it returns `400 This endpoint is only available in SELF_HOSTED mode`. Check `GET /api/config/deployment` → `deployment_mode`.
</Warning>

### Creating a user

`POST /auth/register` creates a user with an email and a password. It only exists when the deployment allows self-registration (otherwise it returns `404`). The server decides the `is_active`, `is_superuser` and `is_verified` flags itself and ignores them in the body.

```bash theme={null}
curl -X POST "$POPSINK_URL/api/auth/register" \
  -H "Content-Type: application/json" \
  -d '{"email": "jane.smith@company.com", "password": "SecurePass456!"}'
```

From now on, every example sends `-H "Authorization: Bearer $TOKEN"`.

***

## Step 2: Pick an environment

List the environments you can reach:

```bash theme={null}
curl "$POPSINK_URL/api/envs/" -H "Authorization: Bearer $TOKEN"
```

```json theme={null}
{
  "items": [
    { "id": "d455a981-be17-455e-a03c-4d9c3224db72", "name": "production" }
  ],
  "total": 1, "page": 1, "size": 50, "pages": 1
}
```

Endpoints that take no `env` / `env_id` parameter use your **active environment**. Set it explicitly:

```bash theme={null}
curl -X PATCH "$POPSINK_URL/api/users/me" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"active_env_id": "d455a981-be17-455e-a03c-4d9c3224db72"}'
```

### Creating an environment

An environment needs the Kafka broker that stores its retention (used for replay and lifecycle management):

```bash theme={null}
curl -X POST "$POPSINK_URL/api/envs/" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "production",
    "retention_configuration": {
      "bootstrap_server": "broker.kafka.svc:9092",
      "security_protocol": "SASL_SSL",
      "sasl_mechanism": "SCRAM-SHA-512",
      "sasl_username": "popsink",
      "sasl_password": "broker-password"
    }
  }'
```

| Field | Type | Required | Description |
| - | - | - | - |
| `name` | string | ✅ | Environment name |
| `retention_configuration.bootstrap_server` | string | ✅ | Broker address |
| `retention_configuration.security_protocol` | string | ✅ | `PLAINTEXT`, `SASL_PLAINTEXT`, `SASL_SSL`, `SSL` |
| `retention_configuration.sasl_mechanism` | string | ✅ | `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`, `GSSAPI`, `OAUTHBEARER` |
| `retention_configuration.sasl_username` | string \| null | ✅ | SASL username (`null` without SASL) |
| `retention_configuration.sasl_password` | string \| null | ✅ | SASL password (`null` without SASL) |
| `retention_configuration.ca_cert` / `cert` / `key` | string | ⬜ | TLS / mutual TLS material |
| `retention_configuration.group_id` | string | ⬜ | Consumer group |

The response never returns the broker credentials.

```
ENV_ID=d455a981-be17-455e-a03c-4d9c3224db72
```

***

## Step 3: Create a team

```bash theme={null}
curl -X POST "$POPSINK_URL/api/teams/" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "data-engineering",
    "description": "Team responsible for the CDC pipelines",
    "env_id": "'"$ENV_ID"'"
  }'
```

**Response (201 Created):**

```json theme={null}
{
  "id": "881e0fce-5caa-4f5c-82c6-5eb883c47100",
  "name": "data-engineering",
  "description": "Team responsible for the CDC pipelines",
  "env_id": "d455a981-be17-455e-a03c-4d9c3224db72"
}
```

Add users as owners (admins of the team) or members. Users who already belong to the team are skipped:

```bash theme={null}
curl -X POST "$POPSINK_URL/api/teams/$TEAM_ID/members/bulk" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "owners": ["3b0c298a-062e-42d1-a01b-277fa66908da"],
    "members": ["32d4b09d-4527-40d5-923b-e166d85cb459"]
  }'
```

**Response:** `204 No Content`. List them with `GET /teams/$TEAM_ID/members`.

```
TEAM_ID=881e0fce-5caa-4f5c-82c6-5eb883c47100
```

***

## Step 4: Check the source credentials

Each connector type has pre-flight endpoints that test a configuration **before** you save it. They take the same body as the connector's `json_configuration` (see [Connector configurations](#connector-configurations)):

```bash theme={null}
curl -X POST "$POPSINK_URL/api/postgres-source/check-credentials" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "host": "pg.company.com",
    "port": 5432,
    "database": "shop",
    "user": "popsink",
    "password": "pg-password"
  }'
```

```json theme={null}
{ "is_success": true, "message": "Credentials are valid." }
```

Then list the tables you can capture:

```bash theme={null}
curl -X POST "$POPSINK_URL/api/postgres-source/list-tables" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{ "host": "pg.company.com", "port": 5432, "database": "shop", "user": "popsink", "password": "pg-password" }'
```

***

## Step 5: Create the source connector

```bash theme={null}
curl -X POST "$POPSINK_URL/api/connectors/" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "shop-postgres",
    "connector_type": "POSTGRES_SOURCE",
    "team_id": "'"$TEAM_ID"'",
    "json_configuration": {
      "host": "pg.company.com",
      "port": 5432,
      "database": "shop",
      "schema": "public",
      "user": "popsink",
      "password": { "valueFrom": { "secretName": "shop-pg", "key": "password" } },
      "whitelist": "public.orders,public.customers"
    }
  }'
```

| Field | Type | Required | Description |
| - | - | - | - |
| `name` | string | ✅ | Connector name |
| `connector_type` | string | ✅ | One of the [connector types](#connector-types) |
| `team_id` | UUID | ✅ | Team that owns the connector |
| `json_configuration` | object | ✅ | Type-specific configuration (see [Connector configurations](#connector-configurations)) |

**Response (201 Created):**

```json theme={null}
{
  "id": "a1b2c3d4-e5f6-4a5b-8c9d-0e1f2a3b4c5d",
  "name": "shop-postgres",
  "connector_type": "POSTGRES_SOURCE",
  "team_id": "881e0fce-5caa-4f5c-82c6-5eb883c47100",
  "json_configuration": {
    "host": "pg.company.com",
    "port": 5432,
    "database": "shop",
    "schema": "public",
    "user": "popsink",
    "password": { "valueFrom": { "secretName": "shop-pg", "key": "password" } },
    "whitelist": "public.orders,public.customers"
  },
  "items_count": 0,
  "status": "paused",
  "topics_config_key": "whitelist",
  "target_destination_kind": null,
  "config_hash": "3f1c..."
}
```

Popsink creates **one data model per entry** of the table list, named `{connector name}_{entry}`. The list is `whitelist` for database sources and `topic` for Kafka (`topics_config_key` tells you which). A CDC source with an empty table list is refused.

```
SOURCE_ID=a1b2c3d4-e5f6-4a5b-8c9d-0e1f2a3b4c5d
```

List the data models it produced:

```bash theme={null}
curl "$POPSINK_URL/api/datamodels/?source_connector_id=$SOURCE_ID" \
  -H "Authorization: Bearer $TOKEN"
```

```json theme={null}
{
  "items": [
    {
      "id": "0c7e6a1e-2b1f-4e7a-9a43-5d2f8b1c9e01",
      "name": "shop-postgres_public.orders",
      "source_name": "shop-postgres",
      "source_schema": "public",
      "source_table": "orders",
      "default_target_table": "orders",
      "status": "paused",
      "enabled": true,
      "subscription_count": 0
    },
    { "id": "7d1f...", "name": "shop-postgres_public.customers", "...": "..." }
  ],
  "total": 2, "page": 1, "size": 50, "pages": 1
}
```

To capture more tables later, `PATCH` the connector with a longer `whitelist`: the new entries become new data models.

***

## Step 6: Create the target connector

Same endpoint, with a target type:

```bash theme={null}
curl -X POST "$POPSINK_URL/api/connectors/" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "analytics-snowflake",
    "connector_type": "SNOWFLAKE_TARGET",
    "team_id": "'"$TEAM_ID"'",
    "json_configuration": {
      "account": "myorg-myaccount",
      "user": "POPSINK",
      "role": "POPSINK_ROLE",
      "warehouse": "POPSINK_WH",
      "database": "ANALYTICS",
      "schema": "RAW",
      "private_key": { "valueFrom": { "secretName": "snowflake", "key": "private-key" } }
    }
  }'
```

```
TARGET_ID=f6e5d4c3-b2a1-4c5d-9e8f-5d4c3b2a1f0e
```

***

## Step 7: Subscribe the data models to the target

A subscription delivers one data model to one target connector:

```bash theme={null}
curl -X POST "$POPSINK_URL/api/subscriptions/" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "orders-to-snowflake",
    "datamodel_id": "0c7e6a1e-2b1f-4e7a-9a43-5d2f8b1c9e01",
    "target_connector_id": "'"$TARGET_ID"'",
    "target_table_name": "ORDERS",
    "backfill": true
  }'
```

**Response (201 Created):**

```json theme={null}
{ "id": "5efd3d63-a6eb-4b0c-afe9-6ee1d8f5f820" }
```

| Field | Type | Required | Description |
| - | - | - | - |
| `name` | string | ✅ | Subscription name |
| `datamodel_id` | UUID | ✅ | Data model to deliver |
| `target_connector_id` | UUID | ✅ | Target connector to write to |
| `target_table_name` | string | ⬜ | Destination table (or topic for a Kafka target). Omit it to use the data model's `default_target_table` |
| `backfill` | boolean | ⬜ | Load the existing rows before streaming changes (default: `false`) |
| `smt_config` | array | ⬜ | Column mapping (see [Mapper configuration](#mapper-configuration)). Omit it to deliver the columns as they are |
| `consumer_id` | string | ⬜ | Consumer group ID (auto-generated if not provided) |
| `error_table_enabled` | boolean | ⬜ | Write rejected records to an error table (default: `false`) |
| `error_table_name` | string | ⬜ | Name of the error table |

To deliver one data model to several targets at once:

```bash theme={null}
curl -X POST "$POPSINK_URL/api/subscriptions/subscribe-all" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "datamodel_id": "0c7e6a1e-2b1f-4e7a-9a43-5d2f8b1c9e01",
    "target_connector_ids": ["f6e5d4c3-b2a1-4c5d-9e8f-5d4c3b2a1f0e", "9a8b7c6d-5e4f-4a3b-8c2d-1e0f9a8b7c6d"]
  }'
```

```json theme={null}
{ "ids": ["5efd3d63-...", "b71c09e2-..."] }
```

### Mapper configuration

`smt_config` is a list of output columns. Each entry:

| Field | Type | Description |
| - | - | - |
| `key` | string | Output column name (required) |
| `path` | string | JSON path of the source value (e.g. `$.customer.id`) |
| `static` | string | Static value, instead of `path` |
| `static_on` | string | `always` or `create`: when the static value is written (default: `always`) |
| `cast` | string | `string`, `int`, `float`, `bool`, `date`, `datetime`, `time` |
| `cast_format` | string | Format used by the cast (e.g. a date format) |
| `primary_key` | boolean | Column is part of the primary key (default: `false`) |
| `nullable` | boolean | Column accepts nulls (default: `true`) |
| `concatenate_fields` | array | Build the value by concatenating several fields |
| `hash_method` | string | Hash the value: `md5`, `sha256`, `sha512` |
| `encrypt_method` / `encrypt_key` | string | Encrypt the value with `aes-256-gcm` |

```json theme={null}
"smt_config": [
  { "key": "order_id", "path": "$.id", "cast": "int", "primary_key": true, "nullable": false },
  { "key": "customer_email", "path": "$.email", "hash_method": "sha256" },
  { "key": "created_at", "path": "$.created_at", "cast": "datetime" },
  { "key": "source", "static": "shop" }
]
```

Test a mapping on a sample message with `POST /smt/process_mapper` (`config`, `table_name`, `message`) before you save it.

***

## Step 8: Start and monitor

### Start the workers

Connectors are created `paused`. Start the source, then the target:

```bash theme={null}
curl -X POST "$POPSINK_URL/api/connectors/$SOURCE_ID/start" -H "Authorization: Bearer $TOKEN"
curl -X POST "$POPSINK_URL/api/connectors/$TARGET_ID/start" -H "Authorization: Bearer $TOKEN"
```

Both return `202 Accepted` before the worker converges. Poll the connector until `status` is `live` (or `error`):

```bash theme={null}
curl "$POPSINK_URL/api/connectors/$SOURCE_ID" -H "Authorization: Bearer $TOKEN" | jq .status
```

| State | Meaning |
| - | - |
| `paused` | No worker running |
| `building` | The worker is starting |
| `live` | The worker is running |
| `degraded` | Some of the connector's worker shards are up, others are down |
| `stopping` | A stop was accepted and the worker is draining |
| `error` | The worker failed; check the logs |

Stop a connector with `POST /connectors/{id}/stop`: it reads `stopping`, then `paused` once the worker is gone.

### Enable or disable a single table

You can disable one data model or one subscription without stopping the connector. Popsink restarts the worker automatically to apply the change:

| Action | Endpoint |
| - | - |
| Stop capturing a table | `POST /datamodels/{id}/stop` |
| Resume capturing a table | `POST /datamodels/{id}/start` |
| Stop delivering to a target | `POST /subscriptions/{id}/pause` |
| Resume delivering to a target | `POST /subscriptions/{id}/start` |

### Monitor

```bash theme={null}
# Worker logs (add ?filter_ids=<datamodel or subscription ids> to narrow them)
curl "$POPSINK_URL/api/connectors/$SOURCE_ID/logs" -H "Authorization: Bearer $TOKEN"

# Data models in error in an environment
curl "$POPSINK_URL/api/datamodels/?env=$ENV_ID&state=error" -H "Authorization: Bearer $TOKEN"

# Throughput and freshness of one data model
curl "$POPSINK_URL/api/datamodels/$DATAMODEL_ID/monitoring" -H "Authorization: Bearer $TOKEN"

# Production metrics of a source over the last day
curl "$POPSINK_URL/api/workers/production-metrics?since=2026-10-01T00:00:00Z&connector_id=$SOURCE_ID" \
  -H "Authorization: Bearer $TOKEN"
```

***

## Step 9: Update a connector or a subscription

### Connector

`PATCH /connectors/{id}` accepts `name`, `json_configuration` and `team_id`. **`json_configuration` is replaced as a whole, not merged**: send the full object, since any key you omit is dropped.

```bash theme={null}
curl -X PATCH "$POPSINK_URL/api/connectors/$SOURCE_ID" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "json_configuration": {
      "host": "pg.company.com",
      "port": 5432,
      "database": "shop",
      "schema": "public",
      "user": "popsink",
      "password": { "valueFrom": { "secretName": "shop-pg", "key": "password" } },
      "whitelist": "public.orders,public.customers,public.invoices"
    }
  }'
```

A `PATCH` never shrinks the table list: entries you omit from `whitelist` / `topic` are kept. To stop capturing a table for good, delete its data model (`DELETE /datamodels/{id}`). `connector_type` can't change.

### Subscription

All fields of `PATCH /subscriptions/{id}` are optional. The target worker restarts to apply the change. The column mapping is called `mapper_config` here:

```bash theme={null}
curl -X PATCH "$POPSINK_URL/api/subscriptions/$SUBSCRIPTION_ID" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "target_table_name": "ORDERS_V2",
    "mapper_config": [
      { "key": "order_id", "path": "$.id", "cast": "int", "primary_key": true, "nullable": false }
    ]
  }'
```

***

## Step 10: Backfill

Reload existing rows into the destinations. Preview first with `POST /backfills/preview`, then trigger:

```bash theme={null}
curl -X POST "$POPSINK_URL/api/backfills/" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{ "datamodel_ids": ["0c7e6a1e-2b1f-4e7a-9a43-5d2f8b1c9e01"] }'
```

You can scope a backfill with `source_connector_ids`, `datamodel_ids` and `target_connector_ids`. Follow it with `GET /backfills/status?source_connector_id=<SOURCE_ID>`, and cancel it with `POST /backfills/{run_id}/cancel`.

<Warning>
  `"truncate_history": true` empties every destination table before reloading it. It is the only option that deletes data in your warehouse.
</Warning>

***

## Connector configurations

The `json_configuration` of a connector depends on its type. The examples below cover the most common ones. `GET /connector-types/` returns every type with its form fields, and the `POST /<type>/check-credentials` body in the [OpenAPI schema](#support) documents each field.

### Secrets

Any sensitive field (passwords, private keys, tokens, service-account JSON) accepts either an inline value or a reference to a Kubernetes secret that already exists in the worker namespace:

```json theme={null}
"password": { "valueFrom": { "secretName": "shop-pg", "key": "password" } }
```

Popsink never stores the referenced value: the worker pod reads it at runtime. We recommend this pattern for API and Terraform consumers. Inline secrets work too, but every read returns them as `"<redacted>"`. `GET /k8s-secrets` lists the secrets available in the worker namespace.

When you re-test an existing connector whose secrets come back as `"<redacted>"`, add `?connector_id=<CONNECTOR_ID>` to its `check-credentials` call: Popsink uses the stored secrets instead of the redacted values.

### SSH tunnel

Database sources (PostgreSQL, MySQL, MariaDB, SQL Server, Oracle, IBM i) can connect through an SSH bastion:

```json theme={null}
"ssh_tunnel": true,
"ssh_host": "bastion.company.com",
"ssh_port": 22,
"ssh_user": "popsink",
"ssh_private_key": { "valueFrom": { "secretName": "bastion", "key": "id_rsa_b64" } }
```

`ssh_private_key` is the **base64-encoded** private key.

***

### KAFKA\_SOURCE Configuration

[Kafka source documentation](/connectors/source/kafka)

```json theme={null}
{
  "bootstrap_servers": "kafka1:9092,kafka2:9092",
  "security_protocol": "SASL_SSL",
  "sasl_mechanism": "SCRAM-SHA-512",
  "sasl_username": "kafka_user",
  "sasl_password": "kafka_pass",
  "topic": "user-events,order-events",
  "consumer_group_id": "pipeline-consumer",
  "include_metadata": false
}
```

| Field | Type | Required | Description |
| - | - | - | - |
| `bootstrap_servers` | string | ✅ | Comma-separated `host:port` brokers |
| `security_protocol` | string | ✅ | `PLAINTEXT`, `SASL_PLAINTEXT`, `SASL_SSL`, `SSL` |
| `sasl_mechanism` | string | ✅ | `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`, `GSSAPI`, `OAUTHBEARER`. Always send it: it is ignored with `PLAINTEXT` or `SSL` |
| `sasl_username` | string \| null | ✅ | SASL username. The key must be present; set it to `null` without SASL |
| `sasl_password` | string \| secret \| null | ✅ | SASL password. The key must be present; set it to `null` without SASL |
| `topic` | string | ✅ | Topics to consume, comma-separated. One data model per topic |
| `consumer_group_id` | string | ⬜ | Consumer group (auto-generated if not provided) |
| `include_metadata` | boolean | ⬜ | Add the Kafka metadata (topic, partition, offset, timestamp) to every event (default: `false`) |
| `ca_cert` | string | ⬜ | CA certificate of the brokers (PEM or base64-encoded PEM) |
| `cert` | string | ⬜ | Client certificate for mutual TLS |
| `key` | string \| secret | ⬜ | Client private key for mutual TLS |

Pre-flight endpoints:

| Endpoint | Response |
| - | - |
| `POST /kafka-source/check-credentials` | `{"is_success": true, "message": "Credentials are valid."}` |
| `POST /kafka-source/list-topics` | `{"topics": ["user-events", "order-events"]}` |
| `POST /kafka-source/fetch-messages` | `{"messages": [ ... ]}` |

`fetch-messages` previews messages. Its body takes exactly one of these shapes: `{"datamodelId": "..."}`, `{"connectorId": "...", "source_topic": "user-events"}`, or the inline configuration above.

***

### POSTGRES\_SOURCE Configuration

[PostgreSQL source documentation](/connectors/source/postgres)

```json theme={null}
{
  "host": "pg.company.com",
  "port": 5432,
  "database": "shop",
  "schema": "public",
  "user": "popsink",
  "password": "pg-password",
  "whitelist": "public.orders,public.customers",
  "publication_name": "popsink_pub",
  "replication_slot_name": "popsink_slot"
}
```

| Field | Type | Required | Description |
| - | - | - | - |
| `host` | string | ✅ | Hostname or IP address |
| `port` | integer | ⬜ | Port (default: `5432`) |
| `database` | string | ✅ | Database name |
| `schema` | string | ⬜ | Schema to list tables from (default: `public`) |
| `user` | string | ✅ | User with replication privileges |
| `password` | string \| secret | ✅ | Password |
| `whitelist` | string | ✅ | Tables to capture, comma-separated, as `schema.table` |
| `publication_name` | string | ⬜ | Publication to use |
| `replication_slot_name` | string | ⬜ | Replication slot to use |
| `ssh_*` | | ⬜ | See [SSH tunnel](#ssh-tunnel) |

Pre-flight: `POST /postgres-source/check-credentials`, `/list-tables`, `/fetch-schema`.

***

### MYSQL\_SOURCE Configuration

[MySQL source documentation](/connectors/source/mysql). `MARIADB_SOURCE` takes the same fields.

```json theme={null}
{
  "host": "mysql.company.com",
  "port": 3306,
  "database": "shop",
  "user": "popsink",
  "password": "mysql-password",
  "whitelist": "shop.orders,shop.customers",
  "encrypt": true,
  "trust_server_certificate": false
}
```

| Field | Type | Required | Description |
| - | - | - | - |
| `host` | string | ✅ | Hostname or IP address |
| `port` | integer | ⬜ | Port (default: `3306`) |
| `database` | string | ✅ | Database (the binlog must be enabled on the server) |
| `user` | string | ✅ | User with replication privileges |
| `password` | string \| secret | ✅ | Password |
| `whitelist` | string | ✅ | Tables to capture, comma-separated, as `database.table` or `table` |
| `encrypt` | boolean | ⬜ | Connect over TLS (default: `true`) |
| `trust_server_certificate` | boolean | ⬜ | Skip the server certificate check (default: `true`) |
| `snapshot_locking_mode` | string | ⬜ | `none` (default) or `minimal` (needs `RELOAD` and `LOCK TABLES`) |
| `ssh_*` | | ⬜ | See [SSH tunnel](#ssh-tunnel) |

Pre-flight: `POST /mysql-source/check-credentials`, `/list-tables`, `/fetch-schema` (`/mariadb-source/...` for MariaDB).

***

### MSSQL\_SOURCE Configuration

[SQL Server source documentation](/connectors/source/mssql)

```json theme={null}
{
  "host": "sqlserver.company.com",
  "port": 1433,
  "database": "Shop",
  "user": "popsink",
  "password": "mssql-password",
  "whitelist": "dbo.Orders,dbo.Customers",
  "encrypt": true,
  "trust_server_certificate": false
}
```

| Field | Type | Required | Description |
| - | - | - | - |
| `host` | string | ✅ | Hostname or IP address |
| `port` | integer | ⬜ | Port (default: `1433`) |
| `database` | string | ✅ | Database (CDC must be enabled on it) |
| `user` | string | ✅ | SQL Server login |
| `password` | string \| secret | ✅ | Password |
| `whitelist` | string | ✅ | Tables to capture, comma-separated, as `schema.table` (`dbo` if omitted) |
| `encrypt` | boolean | ⬜ | Encrypt the JDBC connection (default: `true`) |
| `trust_server_certificate` | boolean | ⬜ | Accept a self-signed certificate (default: `true`) |
| `ssh_*` | | ⬜ | See [SSH tunnel](#ssh-tunnel) |

Pre-flight: `POST /mssql-source/check-credentials`, `/list-tables`, `/fetch-schema`. To read the transaction log without CDC tables, use `MSSQL_LOG_SOURCE` (`/mssql-log-source/...`).

***

### ORACLE\_SOURCE Configuration

[Oracle source documentation](/connectors/source/oracle)

```json theme={null}
{
  "host": "oracle.company.com",
  "port": 1521,
  "service_name": "ORCLCDB",
  "pdb_name": "ORCLPDB1",
  "user": "c##popsink",
  "password": "oracle-password",
  "whitelist": "SALES.ORDERS,SALES.CUSTOMERS"
}
```

| Field | Type | Required | Description |
| - | - | - | - |
| `host` | string | ✅ | Hostname or IP address |
| `port` | integer | ⬜ | Listener port (default: `1521`) |
| `service_name` | string | ✅ or `sid` | Service name. Provide this **or** `sid` |
| `sid` | string | ✅ or `service_name` | SID of a legacy database |
| `pdb_name` | string | ⬜ | Pluggable database to capture from (multitenant only) |
| `user` | string | ✅ | User |
| `password` | string \| secret | ✅ | Password |
| `whitelist` | string | ✅ | Tables to capture, comma-separated, as `SCHEMA.TABLE` |
| `encrypt` | boolean | ⬜ | Connect over TCPS (default: `false`) |
| `wallet` / `wallet_password` | string \| secret | ⬜ | Base64-encoded wallet zip and its password, for mTLS or a private CA |
| `poll_interval_ms` | integer | ⬜ | Polling interval (default: `2000`) |
| `ssh_*` | | ⬜ | See [SSH tunnel](#ssh-tunnel) |

Pre-flight: `POST /oracle-source/check-credentials`, `/list-tables`, `/fetch-schema`.

***

### IBMI\_SOURCE Configuration

[IBM i source documentation](/connectors/source/ibmi)

```json theme={null}
{
  "host": "ibmi.company.com",
  "port": 446,
  "user": "POPSINK",
  "password": "ibmi-password",
  "schemas": ["SALESLIB"],
  "whitelist": "SALESLIB.ORDERS,SALESLIB.CUSTOMERS",
  "char_trim_mode": "both"
}
```

| Field | Type | Required | Description |
| - | - | - | - |
| `host` | string | ✅ | Hostname or IP address |
| `port` | integer | ⬜ | DRDA port (default: `446`) |
| `user` | string | ✅ | User |
| `password` | string \| secret | ✅ | Password |
| `schemas` | array | ✅ | Libraries to capture from. They must all journal to the same journal |
| `whitelist` | string | ✅ | Tables to capture, comma-separated, as `LIB.TABLE` |
| `char_trim_mode` | string | ⬜ | How CHAR/VARCHAR values are trimmed: `both` (default), `leading`, `trailing`, `none` |
| `unicode_escape` | boolean | ⬜ | Escape column names Avro refuses (`ORD#` → `ORD_u0023`, decoded back by the target) (default: `true`) |
| `incremental_load` | boolean | ⬜ | Enable per-table incremental load (default: `false`) |
| `poll_interval_ms` | integer | ⬜ | Polling interval (default: `2000`) |
| `ssh_*` | | ⬜ | See [SSH tunnel](#ssh-tunnel) |

Pre-flight: `POST /ibmi-source/check-credentials`, `/list-tables`, `/fetch-schema`.

***

### MONGO\_SOURCE Configuration

[MongoDB source documentation](/connectors/source/mongodb)

```json theme={null}
{
  "connection_string": "mongodb+srv://cluster0.example.mongodb.net/",
  "database": "shop",
  "user": "popsink",
  "password": "mongo-password",
  "auth_source": "admin",
  "whitelist": "orders,customers",
  "collection_schemas": {
    "orders": { "_id": "objectid", "total": "double", "created_at": "timestamp" },
    "customers": { "_id": "objectid", "email": "string" }
  }
}
```

| Field | Type | Required | Description |
| - | - | - | - |
| `connection_string` | string | ✅ | MongoDB URI **without** credentials (replica set or `mongodb+srv://`) |
| `database` | string | ✅ | Database holding the collections |
| `user` | string | ✅ | User |
| `password` | string \| secret | ✅ | Password |
| `auth_source` | string | ⬜ | Authentication database (default: `admin`) |
| `whitelist` | string | ✅ | Collections to capture, comma-separated |
| `collection_schemas` | object | ⬜ | Columns of each collection: `{collection: {field: type}}`. Types: `string`, `int32`, `int64`, `double`, `bool`, `timestamp`, `binary`, `objectid`, `json`. Every collection needs `_id`; undeclared fields arrive in the `__undeclared` column |

Pre-flight: `POST /mongo-source/check-credentials`, `/list-collections`, `/fetch-schema` (samples a collection to propose its columns).

***

### SNOWFLAKE\_TARGET Configuration

[Snowflake target documentation](/connectors/target/snowflake)

```json theme={null}
{
  "account": "myorg-myaccount",
  "user": "POPSINK",
  "role": "POPSINK_ROLE",
  "warehouse": "POPSINK_WH",
  "database": "ANALYTICS",
  "schema": "RAW",
  "private_key": "MIIEvQIBADANBgkqhkiG9w0BAQEFAASC...",
  "use_dynamic_tables": true
}
```

| Field | Type | Required | Description |
| - | - | - | - |
| `account` | string | ✅ | Account identifier (e.g. `myorg-myaccount`) |
| `user` | string | ✅ | Key-pair user |
| `role` | string | ✅ | Role |
| `database` | string | ✅ | Database |
| `schema` | string | ✅ | Schema |
| `private_key` | string \| secret | ✅ | Private key body, without the `BEGIN` / `END` lines |
| `private_key_passphrase` | string \| secret | ⬜ | Passphrase of an encrypted PKCS#8 key |
| `warehouse` | string | ⬜ | Warehouse used to build the dynamic (merge) tables |
| `use_dynamic_tables` | boolean | ⬜ | Build dynamic (merge) tables; `false` keeps the target append-only (default: `true`) |

Pre-flight: `POST /snowflake-target/check-credentials`, `/warehouses`.

***

### BIGQUERY\_TARGET Configuration

[BigQuery target documentation](/connectors/target/bigquery)

```json theme={null}
{
  "service_account": "{\"type\": \"service_account\", \"project_id\": \"my-project\", ...}",
  "project": "my-project",
  "dataset": "popsink_raw"
}
```

| Field | Type | Required | Description |
| - | - | - | - |
| `service_account` | string \| secret | ✅ | Full content of the service-account JSON key, as a string |
| `dataset` | string | ✅ | Existing dataset (Popsink creates tables, not datasets) |
| `project` | string | ⬜ | Project ID (taken from the key's `project_id` when empty) |

Pre-flight: `POST /bigquery-target/check-credentials`.

***

### POSTGRES\_TARGET Configuration

[PostgreSQL target documentation](/connectors/target/postgres). `MSSQL_TARGET` takes the same fields (default port `1433`, default schema `dbo`).

```json theme={null}
{
  "host": "warehouse.company.com",
  "port": "5432",
  "database": "analytics",
  "schema": "raw",
  "user": "popsink",
  "password": "pg-password"
}
```

| Field | Type | Required | Description |
| - | - | - | - |
| `host` | string | ✅ | Hostname or IP address |
| `port` | string | ⬜ | Port (default: `5432`) |
| `database` | string | ✅ | Database |
| `schema` | string | ⬜ | Target schema (default: `public`) |
| `user` | string | ✅ | User |
| `password` | string \| secret | ✅ | Password |

Pre-flight: `POST /postgres-target/check-credentials` (`/mssql-target/...` for SQL Server).

***

### KAFKA\_TARGET Configuration

[Kafka target documentation](/connectors/target/kafka)

```json theme={null}
{
  "bootstrap_servers": "kafka1:9092,kafka2:9092",
  "security_protocol": "SASL_SSL",
  "sasl_mechanism": "PLAIN",
  "sasl_username": "kafka_user",
  "sasl_password": "kafka_pass"
}
```

| Field | Type | Required | Description |
| - | - | - | - |
| `bootstrap_servers` | string | ✅ | Comma-separated `host:port` brokers |
| `security_protocol` | string | ✅ | `PLAINTEXT`, `SASL_PLAINTEXT`, `SASL_SSL`, `SSL` |
| `sasl_mechanism` | string | ✅ | `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`, `GSSAPI`, `OAUTHBEARER` |
| `sasl_username` | string \| null | ✅ | SASL username (`null` without SASL) |
| `sasl_password` | string \| secret \| null | ✅ | SASL password (`null` without SASL) |

For a Kafka target, the subscription's `target_table_name` is the output topic.

***

### ELASTICSEARCH\_TARGET Configuration

[Elasticsearch target documentation](/connectors/target/elasticsearch)

```json theme={null}
{
  "url": "https://es.company.com:9200",
  "auth_method": "api_key",
  "api_key": "VnVhQ2ZHY0JDZGJrUW0tZTVhT3g6dWkybHAyYXhUTm1zeWFrdzl0dk5udw==",
  "verify_ssl": true
}
```

| Field | Type | Required | Description |
| - | - | - | - |
| `url` | string | ✅ | Server URL |
| `auth_method` | string | ⬜ | `basic` (default), `api_key` or `none` |
| `username` / `password` | string / secret | With `basic` | Basic authentication |
| `api_key` | string \| secret | With `api_key` | Base64-encoded API key |
| `verify_ssl` | boolean | ⬜ | Verify the TLS certificate (default: `true`) |

Pre-flight: `POST /elasticsearch-target/check-credentials`.

***

### WEBHOOK\_TARGET Configuration

[Webhook target documentation](/connectors/target/webhook)

```json theme={null}
{
  "host": "https://example.com/webhook",
  "token": "my-webhook-token",
  "authentication_type": "BEARER",
  "request_method": "POST"
}
```

| Field | Type | Required | Description |
| - | - | - | - |
| `host` | string | ✅ | Webhook URL |
| `token` | string \| secret | ⬜ | Authentication token |
| `authentication_type` | string | ⬜ | `BEARER` (token in the `Authorization` header, default) or `URL` (token as a query parameter) |
| `request_method` | string | ⬜ | `POST` (default) or `GET` |

Pre-flight: `POST /webhook-target/check-credentials`.

***

### Connector types

| Category | Types |
| - | - |
| **Database sources (CDC)** | `POSTGRES_SOURCE`, `MYSQL_SOURCE`, `MARIADB_SOURCE`, `MSSQL_SOURCE`, `MSSQL_LOG_SOURCE`, `ORACLE_SOURCE`, `IBMI_SOURCE`, `MONGO_SOURCE` |
| **Database sources (batch)** | `POSTGRES_DLT_SOURCE`, `MYSQL_DLT_SOURCE`, `MSSQL_DLT_SOURCE`, `ORACLE_DLT_SOURCE`, `IBMI_DLT_SOURCE` |
| **Warehouse sources** | `SNOWFLAKE_SOURCE`, `BIGQUERY_SOURCE` |
| **SaaS sources** | `SALESFORCE_SOURCE`, `HUBSPOT_SOURCE`, `ZENDESK_SOURCE`, `SHOPIFY_SOURCE`, `PIPEDRIVE_SOURCE`, `GOOGLE_ADS_SOURCE`, `FACEBOOK_ADS_SOURCE`, `SHAREPOINT_SOURCE`, `DLT_SOURCE` |
| **Other sources** | `KAFKA_SOURCE`, `FILESYSTEM_DLT_SOURCE`, `DATAGEN_SOURCE`, `POPSINK_SOURCE` |
| **Targets** | `SNOWFLAKE_TARGET`, `BIGQUERY_TARGET`, `POSTGRES_TARGET`, `MSSQL_TARGET`, `ORACLE_TARGET`, `KAFKA_TARGET`, `ICEBERG_TARGET`, `POLARIS_TARGET`, `UNITY_CATALOG_TARGET`, `ELASTICSEARCH_TARGET`, `WEBHOOK_TARGET`, `HUBSPOT_TARGET` |

***

## API reference

The main endpoints, relative to `/api`. See `/api/docs` on your instance for the full list.

### Authentication

| Method | Endpoint | Description |
| - | - | - |
| `POST` | `/auth/jwt/login` | Exchange email + password for an access and a refresh token |
| `POST` | `/auth/jwt/refresh` | Exchange a refresh token for a new pair |
| `POST` | `/auth/jwt/login-from-api-key` | Exchange an organization API key for a 10-minute token (self-hosted) |
| `POST` | `/auth/jwt/logout` | Log out |
| `POST` | `/auth/register` | Create a user (when self-registration is enabled) |
| `POST` | `/auth/forgot-password` / `/auth/reset-password` | Reset a password |

### Users

| Method | Endpoint | Description |
| - | - | - |
| `GET` | `/users/me` | Current user, including `active_env_id` |
| `PATCH` | `/users/me` | Update the current user (e.g. `active_env_id`) |
| `POST` | `/users/me/change-password` | Change your password |
| `POST` | `/users/me/service-account-token` | Create a service-account token |
| `GET` | `/users` | List users (admin only) |
| `GET` / `PATCH` / `DELETE` | `/users/{id}` | Read, update or delete a user |

### Environments and teams

| Method | Endpoint | Description |
| - | - | - |
| `GET` / `POST` | `/envs/` | List or create environments |
| `GET` / `PATCH` / `DELETE` | `/envs/{id}` | Read, update or delete an environment |
| `GET` | `/envs/filter-one?name={name}` | Find an environment by name |
| `GET` | `/envs/members` | List environment members |
| `GET` / `POST` | `/teams/` | List or create teams |
| `GET` / `PATCH` / `DELETE` | `/teams/{id}` | Read, update or delete a team |
| `GET` | `/teams/filter-one?name={name}` | Find a team by name |
| `GET` | `/teams/{id}/members` | List team members |
| `POST` | `/teams/{id}/members/bulk` | Add owners and members |
| `DELETE` | `/teams/{id}/members/{member_id}` | Remove a member |

### Connectors

| Method | Endpoint | Description |
| - | - | - |
| `GET` | `/connector-types/` | Every connector type with its form fields |
| `POST` | `/connectors/` | Create a connector |
| `GET` / `PATCH` / `DELETE` | `/connectors/{id}` | Read, update or delete a connector |
| `GET` | `/connectors/source-config?env_id={id}` | List source connectors |
| `GET` | `/connectors/target-config?env_id={id}` | List target connectors |
| `GET` | `/connectors/filter-one?name={name}` | Find a connector by name |
| `GET` | `/connectors/{id}/source-topics` | Tables or topics available on a source, with stored credentials |
| `POST` | `/connectors/{id}/start` | Start the worker (202) |
| `POST` | `/connectors/{id}/stop` | Stop the worker (202) |
| `GET` | `/connectors/{id}/logs` | Snapshot of the worker logs |
| `POST` | `/connectors/{id}/trigger-incremental-load` | Trigger an incremental load (IBM i) |
| `POST` | `/<type>/check-credentials` | Test a configuration before saving it |
| `POST` | `/<type>/list-tables` | List the tables a source can capture |
| `POST` | `/<type>/fetch-schema` | Fetch the schema of a source table |

### Data models

| Method | Endpoint | Description |
| - | - | - |
| `GET` | `/datamodels/` | List data models (filters: `env`, `team_id`, `source_connector_id`, `state`, `search`) |
| `GET` / `PATCH` / `DELETE` | `/datamodels/{id}` | Read, update or delete a data model |
| `GET` | `/datamodels/count-status` | Count data models by status |
| `GET` | `/datamodels/{id}/schema` | Schema of a data model |
| `GET` | `/datamodels/{id}/monitoring` | Monitoring of a data model |
| `GET` | `/datamodels/{id}/stream` | Recent records of its topic |
| `POST` | `/datamodels/{id}/start` / `stop` | Enable or disable capture of one table (202) |
| `POST` | `/datamodels/{id}/replay-source` | Re-read the source topic from the beginning |
| `PATCH` | `/datamodels/{id}/error-table` | Configure the error table |

### Subscriptions

| Method | Endpoint | Description |
| - | - | - |
| `GET` / `POST` | `/subscriptions/` | List or create subscriptions |
| `POST` | `/subscriptions/subscribe-all` | Subscribe one data model to several targets |
| `GET` / `PATCH` / `DELETE` | `/subscriptions/{id}` | Read, update or delete a subscription |
| `POST` | `/subscriptions/{id}/start` / `pause` | Enable or disable delivery (202) |
| `POST` | `/subscriptions/{id}/replay` | Replay from the broker retention |
| `GET` | `/subscriptions/{id}/target-objects` | Objects a deletion with target cleanup would drop |

### Backfills and transformations

| Method | Endpoint | Description |
| - | - | - |
| `POST` | `/backfills/preview` | Preview what a backfill would reload |
| `POST` | `/backfills/` | Trigger a backfill |
| `GET` | `/backfills/status` | Status of a source connector's backfill |
| `POST` | `/backfills/{run_id}/cancel` | Cancel a backfill run |
| `POST` | `/smt/process_mapper` | Apply a mapper configuration to a sample message |

### Monitoring and health

| Method | Endpoint | Description |
| - | - | - |
| `GET` | `/workers/production-metrics` | CDC production metrics since a date |
| `GET` | `/workers/consumption-metrics` | Per-subscription consumption metrics since a date |
| `GET` | `/workers/resource/{id}/status` | Current worker status of a resource |
| `GET` | `/datamodels/environment/monitoring-summary` | Environment monitoring summary |
| `GET` | `/events` | Monitoring events of a resource |
| `GET` | `/config/deployment` | Deployment ID, mode and name |
| `GET` | `/healthchecks` | API health (also `/db`, `/broker`, `/schema-registry`, `/k8s`) |

***

## Troubleshooting

### `401 Unauthorized`

The token is missing or expired. Log in again, refresh it with `POST /auth/jwt/refresh`, or, with an API key, exchange the key again (its tokens last 10 minutes).

### `400 This endpoint is only available in SELF_HOSTED mode`

API-key login is only available on self-hosted deployments. Use email and password login, or check `GET /config/deployment` to confirm which instance you are calling.

### `403 Forbidden`

You are not a member of the team that owns the resource, or your role doesn't allow writes. Ask a team owner to add you with `POST /teams/{id}/members/bulk`.

### `404` on `/auth/register`

Self-registration is disabled on this deployment. Ask an administrator to invite you.

### `409 Conflict`

A connector with the same name already exists. Pick another name, or send an `Idempotency-Key` header so that retries return the original resource.

### `422 Validation Error`

The body doesn't match the schema. The response names the offending field:

```json theme={null}
{
  "detail": [
    { "loc": ["body", "json_configuration"], "msg": "Field required", "type": "missing" }
  ]
}
```

### Credentials come back as `"<redacted>"`

That's expected: inline secrets are redacted on every read. Send the full value (or a `valueFrom` reference) again when you `PATCH` the connector, since `json_configuration` is replaced as a whole.

### Connector stuck in `building`

The worker can't start. Read `GET /connectors/{id}/logs`, then check the configuration with the type's `check-credentials` endpoint.

***

## Complete example script

This script creates a PostgreSQL source and a Snowflake target, subscribes every table to Snowflake and starts both workers:

```python theme={null}
#!/usr/bin/env python3
"""Replicate PostgreSQL tables to Snowflake through the Popsink API."""
import time

import requests

SERVER = "https://<tenant>.<region>.popsink.com/api"
EMAIL = "john.doe@company.com"
PASSWORD = "MySecurePassword123!"
TEAM_ID = "881e0fce-5caa-4f5c-82c6-5eb883c47100"


class PopsinkClient:
    def __init__(self, server: str):
        self.server = server.rstrip("/")
        self.session = requests.Session()

    def login(self, email: str, password: str) -> None:
        response = self.session.post(
            f"{self.server}/auth/jwt/login",
            data={"username": email, "password": password},
        )
        response.raise_for_status()
        token = response.json()["access_token"]
        self.session.headers["Authorization"] = f"Bearer {token}"

    def request(self, method: str, path: str, **kwargs):
        response = self.session.request(method, f"{self.server}{path}", **kwargs)
        response.raise_for_status()
        return response.json() if response.content else None

    def create_connector(self, name: str, connector_type: str, config: dict) -> dict:
        return self.request("POST", "/connectors/", json={
            "name": name,
            "connector_type": connector_type,
            "team_id": TEAM_ID,
            "json_configuration": config,
        })

    def list_datamodels(self, source_id: str) -> list[dict]:
        items, page = [], 1
        while True:
            data = self.request("GET", "/datamodels/", params={
                "source_connector_id": source_id, "page": page, "size": 100,
            })
            items += data["items"]
            if page >= (data["pages"] or 1):
                return items
            page += 1

    def wait_until_live(self, connector_id: str, timeout: int = 600) -> str:
        deadline = time.time() + timeout
        while time.time() < deadline:
            status = self.request("GET", f"/connectors/{connector_id}")["status"]
            if status in ("live", "error"):
                return status
            time.sleep(10)
        raise TimeoutError(f"Connector {connector_id} is still not live")


def main():
    client = PopsinkClient(SERVER)
    client.login(EMAIL, PASSWORD)

    source = client.create_connector("shop-postgres", "POSTGRES_SOURCE", {
        "host": "pg.company.com",
        "port": 5432,
        "database": "shop",
        "user": "popsink",
        "password": {"valueFrom": {"secretName": "shop-pg", "key": "password"}},
        "whitelist": "public.orders,public.customers",
    })
    print(f"Source created: {source['id']}")

    target = client.create_connector("analytics-snowflake", "SNOWFLAKE_TARGET", {
        "account": "myorg-myaccount",
        "user": "POPSINK",
        "role": "POPSINK_ROLE",
        "warehouse": "POPSINK_WH",
        "database": "ANALYTICS",
        "schema": "RAW",
        "private_key": {"valueFrom": {"secretName": "snowflake", "key": "private-key"}},
    })
    print(f"Target created: {target['id']}")

    for datamodel in client.list_datamodels(source["id"]):
        subscription = client.request("POST", "/subscriptions/", json={
            "name": f"{datamodel['source_table'] or datamodel['id']}-to-snowflake",
            "datamodel_id": datamodel["id"],
            "target_connector_id": target["id"],
            "backfill": True,
        })
        print(f"Subscribed {datamodel['name']}: {subscription['id']}")

    for connector in (source, target):
        client.request("POST", f"/connectors/{connector['id']}/start")
        print(f"{connector['name']}: {client.wait_until_live(connector['id'])}")


if __name__ == "__main__":
    main()
```

***

## Support

* **Swagger UI:** `https://<your-instance>/api/docs`
* **OpenAPI spec:** `https://<your-instance>/api/openapi.json`
* **Claude Code skill:** [drive the API from Claude Code](/api/skills)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.