Docs/API/Streams API

Streams API Workflows

Use this page when you need to configure and run data streams through the API.

A stream configuration defines what data to move and how — which source connections, tables, target, and write behavior. Starting a configuration creates a stream execution that can be paused, resumed, or stopped.

Authentication

All endpoints require:

-H "X-API-Key: $API_KEY" \
-H "X-Install-ID: $INSTALL_ID"

Endpoints overview

Stream configurations

RouteMethodDescription
/stream-configsPOSTCreate a configuration
/stream-configsGETList all configurations
/stream-configs/{id}GETGet configuration details
/stream-configs/{id}PUTUpdate a configuration
/stream-configs/{id}DELETEDelete a configuration
/stream-configs/{id}/clonePUTClone a configuration
/stream-configs/{id}/startPOSTStart a stream
/stream-configs/{id}/historyGETGet run history

Stream execution

RouteMethodDescription
/streams/{id}/pausePOSTPause a running stream
/streams/{id}/resumePOSTResume a paused stream
/streams/{id}/stopPOSTStop a stream
/streams/{id}/statsGETGet stream statistics
/streams/{id}/logsGETGet persisted stream logs

Note: /stream-configs/ routes use configuration IDs (config_...). /streams/ routes use stream execution IDs (stream_...).

Understanding IDs

ID prefixExampleUsed with
conn_conn_2sZyDMeIbdaXnN9OL7vYzzAiZU5Connection endpoints
config_config_3B54E8EctFDkIfPiNTqhIogCp5Z/stream-configs/
stream_stream_3B54FGHiJklMnOpQrStUvWxYz12/streams/

Starting a configuration creates a new stream_ ID for that execution. The configuration remains unchanged and can be used to start multiple streams.

Create a stream configuration

POST /stream-configs

Database to database (load mode; API value: load)

curl -X POST "$API_URL/stream-configs" \
  -H "X-API-Key: $API_KEY" \
  -H "X-Install-ID: $INSTALL_ID" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "mysql_to_postgresql",
    "mode": "load",
    "source": {
      "connections": [
        {
          "connectionId": "conn_source_id",
          "database": "sakila",
          "tables": [
            {"name": "actor"},
            {"name": "film"}
          ]
        }
      ]
    },
    "target": {
      "id": "conn_target_id",
      "spec": {
        "db": {
          "database": "postgres",
          "writeMode": "truncate_and_load",
          "structureOptions": {
            "tables": true,
            "indexes": true,
            "foreignKeys": true
          }
        }
      }
    }
  }'

Database to database (CDC mode)

{
  "name": "mysql_cdc_to_pg",
  "mode": "cdc",
  "source": {
    "connections": [
      {
        "connectionId": "conn_source_id",
        "database": "production",
        "tables": [
          {"name": "orders"},
          {"name": "customers"}
        ]
      }
    ],
    "options": {
      "operations": ["INSERT", "UPDATE", "DELETE"]
    }
  },
  "target": {
    "id": "conn_target_id",
    "spec": {
      "db": {
        "database": "replica",
        "writeMode": "upsert",
        "schemaPolicy": "validate_existing"
      }
    }
  }
}

Database to S3

{
  "name": "mysql_to_s3_parquet",
  "mode": "load",
  "source": {
    "connections": [
      {
        "connectionId": "conn_mysql_id",
        "database": "analytics",
        "tables": [{"name": "events"}]
      }
    ]
  },
  "target": {
    "id": "conn_s3_id",
    "spec": {
      "s3": {
        "fileFormat": "parquet",
        "upload": {
          "bucket": "data-lake",
          "prefix": "exports/"
        }
      }
    }
  }
}

Database to local files

{
  "name": "pg_to_csv",
  "mode": "load",
  "source": {
    "connections": [
      {
        "connectionId": "conn_pg_id",
        "database": "reporting",
        "tables": [{"name": "monthly_report"}]
      }
    ]
  },
  "target": {
    "id": "conn_files_id",
    "spec": {
      "files": {
        "fileFormat": "csv",
        "format": {
          "compression": "gzip"
        }
      }
    }
  }
}

Response

Returns the created configuration with a generated config_ ID:

{
  "id": "config_3B54E8EctFDkIfPiNTqhIogCp5Z",
  "name": "mysql_to_postgresql",
  "mode": "load",
  "source": {
    "connections": [
      {
        "connectionId": "conn_source_id",
        "database": "sakila",
        "tables": [
          {"name": "actor"},
          {"name": "film"}
        ]
      }
    ]
  },
  "target": {
    "id": "conn_target_id",
    "spec": {
      "db": {
        "database": "postgres",
        "writeMode": "truncate_and_load",
        "structureOptions": {
          "tables": true,
          "indexes": true,
          "foreignKeys": true
        }
      }
    }
  },
  "created": 1773765482,
  "reportingInterval": 3
}

Configuration fields

Source

{
  "source": {
    "connections": [{...}],
    "options": {...}
  }
}

Each entry in connections specifies:

FieldTypeDescription
connectionIdstringConnection ID (required)
databasestringDatabase name
schemastringSchema name (optional, PostgreSQL/Snowflake)
tablesarrayTables to replicate
queriesarrayVirtual query sources (load mode only; API mode=load)
s3objectS3 source config (mutually exclusive with tables)
filesobjectFile source config (mutually exclusive with tables)

Source options:

FieldTypeDescription
dataBundleSizenumberRows per NATS message (default: 500)
operationsarrayCDC operations filter: INSERT, UPDATE, DELETE
replicationSlotstringPostgreSQL CDC replication slot name (default: {database}_dbconvert_replication_slot)
publicationNamestringPostgreSQL CDC publication name (default: dbconvert-publication)
initialSnapshotobjectCDC-only bootstrap option. Set enabled: true to copy existing rows before CDC starts.

source.options.initialSnapshot.restartPolicy currently supports fail_until_reset.

Table selection

Tables support optional structured selection for filtering and column selection:

{
  "name": "orders",
  "selection": {
    "selectedColumns": ["id", "customer_id", "total"],
    "filters": [
      {"column": "status", "operator": "=", "value": "active"}
    ],
    "sorts": [
      {"column": "created_at", "direction": "DESC"}
    ],
    "limit": 10000
  }
}

Target

{
  "target": {
    "id": "conn_target_id",
    "spec": {
      "db": {...}
    }
  }
}

target.id is the connection ID. target.spec uses one variant matching the target type:

Spec fieldTarget type
dbMySQL, PostgreSQL, Snowflake (data)
filesLocal file system
s3Amazon S3
gcsGoogle Cloud Storage
azureAzure Blob Storage
snowflakeSnowflake (with staging)

Database target spec (db)

FieldTypeDescription
databasestringTarget database name (required)
schemastringTarget schema (optional)
writeModestringHow to handle existing data
schemaPolicystringHow to handle existing schema
structureOptionsobjectWhich DDL to create (tables, indexes, etc.)
skipDatabooleanCreate structure only, skip data transfer

Write modes:

ValueDescription
fail_if_not_emptyError if target table has data (default for load mode, API mode=load)
appendAdd rows to existing data
truncate_and_loadClear target tables before writing
upsertInsert or update on conflict (default for CDC)

Schema policies:

ValueDescription
fail_if_existsError if target tables exist (default for load mode, API mode=load)
validate_existingUse existing tables, validate schema (default for CDC)
create_missing_onlyCreate only missing tables
drop_and_recreateDrop and recreate all target tables

Structure options:

FieldTypeDescription
tablesbooleanCreate tables
indexesbooleanCreate indexes
foreignKeysbooleanCreate foreign key constraints
checkConstraintsbooleanCreate check constraints

File/S3 target spec

FieldTypeDescription
fileFormatstringcsv, json, jsonl, or parquet
formatobjectFormat-specific settings (compression, etc.)
outputDirectorystringLocal staging path (optional)

S3 targets add an upload object:

{
  "upload": {
    "bucket": "my-bucket",
    "prefix": "exports/",
    "storageClass": "STANDARD"
  }
}

Other configuration fields

FieldTypeDescription
namestringConfiguration name (required)
modestringconvert or cdc (required)
reportingIntervalnumberStats reporting interval in seconds
limitsobjectOptional execution limits
descriptionstringOptional description

Limits:

FieldTypeDescription
numberOfEventsnumberMax rows to process (0 = unlimited)
elapsedTimenumberMax seconds to run (0 = unlimited)

List stream configurations

GET /stream-configs

Returns a wrapped response. Supports search, limit, and offset query parameters.

curl "$API_URL/stream-configs" \
  -H "X-API-Key: $API_KEY" \
  -H "X-Install-ID: $INSTALL_ID"

Response:

{
  "items": [...],
  "total": 93,
  "filtered": 93
}

Get, update, delete

  • GET /stream-configs/{id} — returns the full configuration object
  • PUT /stream-configs/{id} — updates a configuration, returns the updated object
  • DELETE /stream-configs/{id} — requires X-Confirm-Delete: true header, returns 204

Clone a configuration

PUT /stream-configs/{id}/clone

Creates a copy with a new ID and _copy appended to the name. Returns the full cloned configuration.

Start a stream

POST /stream-configs/{id}/start

Starts a stream execution from a configuration. Only one stream can be active at a time.

Optional query parameter: ?structureOnly=true to create target structure without transferring data.

Response:

{
  "id": "stream_3B54FGHiJklMnOpQrStUvWxYz12"
}

Use the returned stream_ ID for pause, resume, stop, and stats endpoints.

Pause, resume, stop

All return HTTP 204 No Content with no response body.

# Pause a running stream
curl -X POST "$API_URL/streams/$STREAM_ID/pause" \
  -H "X-API-Key: $API_KEY" \
  -H "X-Install-ID: $INSTALL_ID"

# Resume a paused stream
curl -X POST "$API_URL/streams/$STREAM_ID/resume" \
  -H "X-API-Key: $API_KEY" \
  -H "X-Install-ID: $INSTALL_ID"

# Stop a stream
curl -X POST "$API_URL/streams/$STREAM_ID/stop" \
  -H "X-API-Key: $API_KEY" \
  -H "X-Install-ID: $INSTALL_ID"

Stop accepts an optional ?force=true query parameter for recovery scenarios.

Get stream statistics

GET /streams/{id}/stats

Returns current execution state and per-node counters. For CDC and chunked snapshot workflows, the same response can also include reliability, bootstrap, and snapshotCopy blocks. See Logs And Monitoring for the full response structure and status values.

curl "$API_URL/streams/$STREAM_ID/stats" \
  -H "X-API-Key: $API_KEY" \
  -H "X-Install-ID: $INSTALL_ID"

Response:

{
  "configID": "config_3B54E8EctFDkIfPiNTqhIogCp5Z",
  "streamID": "stream_3B54FGHiJklMnOpQrStUvWxYz12",
  "status": "RUNNING",
  "nodes": [
    {
      "id": "reader-1",
      "type": "source",
      "stat": {
        "counter": 15000,
        "failedCounter": 0,
        "sumDataSize": 2048576,
        "avgRate": 17067,
        "status": "RUNNING"
      }
    }
  ]
}

When present, the optional workflow blocks look like this:

{
  "bootstrap": {
    "engine": "mysql",
    "phase": "copying",
    "handoffPosition": "mysql-bin.000176:79269"
  },
  "snapshotCopy": {
    "status": "copying",
    "mode": "cdc_bootstrap",
    "strategy": "pk_range",
    "tablesPlanned": 12,
    "tablesCompleted": 4,
    "chunksPlanned": 96,
    "chunksCompleted": 31
  }
}
  • bootstrap appears when source.options.initialSnapshot.enabled = true and bootstrap state has been persisted.
  • snapshotCopy appears when load mode (API mode=load) or MySQL CDC bootstrap is actively using resumable chunked snapshot copy for at least one eligible table.

Stream states

StatusDescription
RUNNINGActively processing data
PAUSEDTemporarily paused, can be resumed
FINISHEDCompleted successfully
FAILEDTerminated with an error
STOPPEDManually stopped
TIME_LIMIT_REACHEDStopped after reaching time limit
EVENT_LIMIT_REACHEDStopped after reaching event limit

Error handling

{
  "error": "Cannot start new stream: found active streams that need attention",
  "type": "validation",
  "code": "VALIDATION_FAILED",
  "details": {
    "field": "stream_state",
    "reason": "Cannot start new stream: found active streams that need attention"
  }
}

Common status codes:

CodeMeaning
400Validation error (missing fields, invalid config)
401Invalid or missing API key
404Configuration or stream not found
503Service error