Stream API Code Examples

# Set credentials
export API_KEY="your_api_key_here"
export INSTALL_ID="your_install_id_here"
export API_URL="http://127.0.0.1:8020/api/v1"

# Load user configurations (required before other endpoints)
curl "$API_URL/user/configs" \
  -H "X-API-Key: $API_KEY" \
  -H "X-Install-ID: $INSTALL_ID"

# Create a stream configuration
config_result=$(curl -s -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
          }
        }
      }
    }
  }')

# Extract configuration ID
CONFIG_ID=$(echo $config_result | jq -r '.id')
echo "Config ID: $CONFIG_ID"  # config_...

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

# Start a stream from the configuration
stream_result=$(curl -s -X POST "$API_URL/stream-configs/$CONFIG_ID/start" \
  -H "X-API-Key: $API_KEY" \
  -H "X-Install-ID: $INSTALL_ID")

# Extract stream execution ID
STREAM_ID=$(echo $stream_result | jq -r '.id')
echo "Stream ID: $STREAM_ID"  # stream_...

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

# Pause a running stream (returns 204 No Content)
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 (returns 204 No Content)
curl -X POST "$API_URL/streams/$STREAM_ID/resume" \
  -H "X-API-Key: $API_KEY" \
  -H "X-Install-ID: $INSTALL_ID"

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

# Get run history for a configuration
curl "$API_URL/stream-configs/$CONFIG_ID/history" \
  -H "X-API-Key: $API_KEY" \
  -H "X-Install-ID: $INSTALL_ID"

# Get persisted logs for a completed stream
curl "$API_URL/streams/$STREAM_ID/logs" \
  -H "X-API-Key: $API_KEY" \
  -H "X-Install-ID: $INSTALL_ID"

# Delete a configuration (requires confirmation header)
curl -X DELETE "$API_URL/stream-configs/$CONFIG_ID" \
  -H "X-API-Key: $API_KEY" \
  -H "X-Install-ID: $INSTALL_ID" \
  -H "X-Confirm-Delete: true"

import requests


class DBConvertStreamsAPI:
    def __init__(self, api_key, install_id, base_url="http://127.0.0.1:8020/api/v1"):
        self.base_url = base_url
        self.headers = {
            "X-API-Key": api_key,
            "X-Install-ID": install_id,
            "Content-Type": "application/json",
        }

    def load_user_configs(self):
        """Load user configurations (required before other endpoints)."""
        response = requests.get(f"{self.base_url}/user/configs", headers=self.headers)
        response.raise_for_status()
        return response.json()

    def create_stream_config(self, config_data):
        """Create a stream configuration. Returns the full config with generated ID."""
        response = requests.post(
            f"{self.base_url}/stream-configs", headers=self.headers, json=config_data
        )
        response.raise_for_status()
        return response.json()

    def list_stream_configs(self):
        """List all configurations. Returns {items, total, filtered}."""
        response = requests.get(
            f"{self.base_url}/stream-configs", headers=self.headers
        )
        response.raise_for_status()
        return response.json()

    def start_stream(self, config_id, structure_only=False):
        """Start a stream from a configuration. Returns {id} with stream_ ID."""
        params = {}
        if structure_only:
            params["structureOnly"] = "true"
        response = requests.post(
            f"{self.base_url}/stream-configs/{config_id}/start",
            headers=self.headers,
            params=params,
        )
        response.raise_for_status()
        return response.json()

    def get_stats(self, stream_id):
        """Get stream statistics. Returns {configID, streamID, status, nodes}."""
        response = requests.get(
            f"{self.base_url}/streams/{stream_id}/stats", headers=self.headers
        )
        response.raise_for_status()
        return response.json()

    def pause_stream(self, stream_id):
        """Pause a running stream. Returns 204 No Content."""
        response = requests.post(
            f"{self.base_url}/streams/{stream_id}/pause", headers=self.headers
        )
        response.raise_for_status()

    def resume_stream(self, stream_id):
        """Resume a paused stream. Returns 204 No Content."""
        response = requests.post(
            f"{self.base_url}/streams/{stream_id}/resume", headers=self.headers
        )
        response.raise_for_status()

    def stop_stream(self, stream_id):
        """Stop a stream. Returns 204 No Content."""
        response = requests.post(
            f"{self.base_url}/streams/{stream_id}/stop", headers=self.headers
        )
        response.raise_for_status()

    def get_history(self, config_id):
        """Get run history for a configuration."""
        response = requests.get(
            f"{self.base_url}/stream-configs/{config_id}/history",
            headers=self.headers,
        )
        response.raise_for_status()
        return response.json()

    def get_logs(self, stream_id):
        """Get persisted logs for a stream run."""
        response = requests.get(
            f"{self.base_url}/streams/{stream_id}/logs", headers=self.headers
        )
        response.raise_for_status()
        return response.json()


# Usage
api = DBConvertStreamsAPI("your_api_key", "your_install_id")
api.load_user_configs()

# Create a stream configuration
config = api.create_stream_config({
    "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,
                },
            }
        },
    },
})
print(f"Config ID: {config['id']}")

# Start a stream
result = api.start_stream(config["id"])
stream_id = result["id"]
print(f"Stream ID: {stream_id}")

# Monitor statistics
stats = api.get_stats(stream_id)
print(f"Status: {stats['status']}")
for node in stats["nodes"]:
    print(f"  {node['id']}: {node['stat']['counter']} rows")

# Pause and resume
api.pause_stream(stream_id)
print("Stream paused")

api.resume_stream(stream_id)
print("Stream resumed")

# Stop
api.stop_stream(stream_id)
print("Stream stopped")

import axios, { AxiosInstance } from 'axios'

interface StreamConfigRequest {
  name: string
  mode: 'load' | 'cdc'
  source: {
    connections: Array<{
      connectionId: string
      database: string
      tables?: Array<{ name: string }>
    }>
    options?: {
      dataBundleSize?: number
      operations?: Array<'INSERT' | 'UPDATE' | 'DELETE'>
    }
  }
  target: {
    id: string
    spec: {
      db?: {
        database: string
        writeMode?: 'fail_if_not_empty' | 'append' | 'truncate_and_load' | 'upsert'
        schemaPolicy?: string
        structureOptions?: {
          tables?: boolean
          indexes?: boolean
          foreignKeys?: boolean
        }
      }
      s3?: {
        fileFormat: 'csv' | 'json' | 'jsonl' | 'parquet'
        upload: { bucket: string; prefix?: string }
      }
      files?: {
        fileFormat: 'csv' | 'json' | 'jsonl' | 'parquet'
      }
    }
  }
  reportingInterval?: number
  limits?: {
    numberOfEvents?: number
    elapsedTime?: number
  }
}

interface StartStreamResponse {
  id: string
}

interface StatResponse {
  configID: string
  streamID: string
  status: string
  nodes: Array<{
    id: string
    type: string
    stat: {
      counter: number
      failedCounter: number
      sumDataSize: number
      avgRate: number
      status: string
    }
  }>
}

interface StreamConfigListResponse {
  items: StreamConfigRequest[]
  total: number
  filtered: number
}

class DBConvertStreamsAPI {
  private client: AxiosInstance

  constructor(apiKey: string, installId: string, baseURL = 'http://127.0.0.1:8020/api/v1') {
    this.client = axios.create({
      baseURL,
      headers: {
        'X-API-Key': apiKey,
        'X-Install-ID': installId,
        'Content-Type': 'application/json',
      },
    })
  }

  async loadUserConfigs(): Promise<void> {
    await this.client.get('/user/configs')
  }

  async createStreamConfig(config: StreamConfigRequest): Promise<StreamConfigRequest & { id: string }> {
    const { data } = await this.client.post('/stream-configs', config)
    return data
  }

  async listStreamConfigs(): Promise<StreamConfigListResponse> {
    const { data } = await this.client.get('/stream-configs')
    return data
  }

  async startStream(configId: string): Promise<StartStreamResponse> {
    const { data } = await this.client.post(`/stream-configs/${configId}/start`)
    return data
  }

  async getStats(streamId: string): Promise<StatResponse> {
    const { data } = await this.client.get(`/streams/${streamId}/stats`)
    return data
  }

  async pauseStream(streamId: string): Promise<void> {
    await this.client.post(`/streams/${streamId}/pause`)
  }

  async resumeStream(streamId: string): Promise<void> {
    await this.client.post(`/streams/${streamId}/resume`)
  }

  async stopStream(streamId: string): Promise<void> {
    await this.client.post(`/streams/${streamId}/stop`)
  }
}

// Usage
async function main() {
  const api = new DBConvertStreamsAPI('your_api_key', 'your_install_id')
  await api.loadUserConfigs()

  // Create a stream configuration
  const config = await api.createStreamConfig({
    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 },
        },
      },
    },
  })
  console.log('Config ID:', config.id)

  // Start a stream
  const { id: streamId } = await api.startStream(config.id)
  console.log('Stream ID:', streamId)

  // Monitor statistics
  const stats = await api.getStats(streamId)
  console.log('Status:', stats.status)
  for (const node of stats.nodes) {
    console.log(`  ${node.id}: ${node.stat.counter} rows`)
  }

  // Pause, resume, stop
  await api.pauseStream(streamId)
  console.log('Stream paused')

  await api.resumeStream(streamId)
  console.log('Stream resumed')

  await api.stopStream(streamId)
  console.log('Stream stopped')
}

main().catch(console.error)

class DBConvertStreamsAPI {
    [string]$BaseURL
    [hashtable]$Headers

    DBConvertStreamsAPI([string]$ApiKey, [string]$InstallId,
                       [string]$BaseURL = "http://127.0.0.1:8020/api/v1") {
        $this.BaseURL = $BaseURL
        $this.Headers = @{
            "X-API-Key"    = $ApiKey
            "X-Install-ID" = $InstallId
            "Content-Type" = "application/json"
        }
    }

    [void] LoadUserConfigs() {
        Invoke-RestMethod -Uri "$($this.BaseURL)/user/configs" `
            -Headers $this.Headers -Method Get
    }

    [object] CreateStreamConfig([hashtable]$Data) {
        $body = $Data | ConvertTo-Json -Depth 10
        return Invoke-RestMethod -Uri "$($this.BaseURL)/stream-configs" `
            -Headers $this.Headers -Method Post -Body $body
    }

    [object] ListStreamConfigs() {
        return Invoke-RestMethod -Uri "$($this.BaseURL)/stream-configs" `
            -Headers $this.Headers -Method Get
    }

    [object] StartStream([string]$ConfigId) {
        return Invoke-RestMethod `
            -Uri "$($this.BaseURL)/stream-configs/$ConfigId/start" `
            -Headers $this.Headers -Method Post
    }

    [object] GetStats([string]$StreamId) {
        return Invoke-RestMethod `
            -Uri "$($this.BaseURL)/streams/$StreamId/stats" `
            -Headers $this.Headers -Method Get
    }

    [void] PauseStream([string]$StreamId) {
        Invoke-RestMethod -Uri "$($this.BaseURL)/streams/$StreamId/pause" `
            -Headers $this.Headers -Method Post
    }

    [void] ResumeStream([string]$StreamId) {
        Invoke-RestMethod -Uri "$($this.BaseURL)/streams/$StreamId/resume" `
            -Headers $this.Headers -Method Post
    }

    [void] StopStream([string]$StreamId) {
        Invoke-RestMethod -Uri "$($this.BaseURL)/streams/$StreamId/stop" `
            -Headers $this.Headers -Method Post
    }
}

# Usage
$api = [DBConvertStreamsAPI]::new("your_api_key", "your_install_id")
$api.LoadUserConfigs()

# Create a stream configuration
$config = $api.CreateStreamConfig(@{
    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
                }
            }
        }
    }
})
Write-Host "Config ID: $($config.id)"

# Start a stream
$result = $api.StartStream($config.id)
$streamId = $result.id
Write-Host "Stream ID: $streamId"

# Monitor statistics
$stats = $api.GetStats($streamId)
Write-Host "Status: $($stats.status)"
foreach ($node in $stats.nodes) {
    Write-Host "  $($node.id): $($node.stat.counter) rows"
}

# Pause, resume, stop
$api.PauseStream($streamId)
Write-Host "Stream paused"

$api.ResumeStream($streamId)
Write-Host "Stream resumed"

$api.StopStream($streamId)
Write-Host "Stream stopped"