Connection 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"

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

# Create a MySQL connection
curl -X POST "$API_URL/connections" \
  -H "X-API-Key: $API_KEY" \
  -H "X-Install-ID: $INSTALL_ID" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "mysql-source",
    "type": "mysql",
    "spec": {
      "database": {
        "host": "localhost",
        "port": 3306,
        "username": "root",
        "password": "your_password",
        "database": "source_db"
      }
    }
  }'

# Create a PostgreSQL connection
curl -X POST "$API_URL/connections" \
  -H "X-API-Key: $API_KEY" \
  -H "X-Install-ID: $INSTALL_ID" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "pg-target",
    "type": "postgresql",
    "spec": {
      "database": {
        "host": "localhost",
        "port": 5432,
        "username": "postgres",
        "password": "your_password",
        "database": "target_db"
      }
    }
  }'

# Test a saved connection
curl -X POST "$API_URL/connections/$CONN_ID/ping" \
  -H "X-API-Key: $API_KEY" \
  -H "X-Install-ID: $INSTALL_ID"

# List databases
curl "$API_URL/connections/$CONN_ID/databases" \
  -H "X-API-Key: $API_KEY" \
  -H "X-Install-ID: $INSTALL_ID"

# List tables in a database
curl "$API_URL/connections/$CONN_ID/databases/source_db/tables" \
  -H "X-API-Key: $API_KEY" \
  -H "X-Install-ID: $INSTALL_ID"

# Delete a connection (requires confirmation header)
curl -X DELETE "$API_URL/connections/$CONN_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 list_connections(self):
        """List all connections. Returns {items, total, filtered}."""
        response = requests.get(f"{self.base_url}/connections", headers=self.headers)
        response.raise_for_status()
        return response.json()

    def create_connection(self, connection_data):
        """Create a connection. Returns {id, created}."""
        response = requests.post(
            f"{self.base_url}/connections", headers=self.headers, json=connection_data
        )
        response.raise_for_status()
        return response.json()

    def ping(self, conn_id):
        """Test a saved connection. Returns {ping, error?}."""
        response = requests.post(
            f"{self.base_url}/connections/{conn_id}/ping", headers=self.headers
        )
        response.raise_for_status()
        return response.json()

    def list_databases(self, conn_id):
        """List databases for a connection."""
        response = requests.get(
            f"{self.base_url}/connections/{conn_id}/databases", headers=self.headers
        )
        response.raise_for_status()
        return response.json()

    def list_tables(self, conn_id, database, schemas=None):
        """List tables in a database. Returns string array."""
        params = {}
        if schemas:
            params["schemas"] = schemas
        response = requests.get(
            f"{self.base_url}/connections/{conn_id}/databases/{database}/tables",
            headers=self.headers,
            params=params,
        )
        response.raise_for_status()
        return response.json()

    def delete_connection(self, conn_id):
        """Delete a connection. Requires X-Confirm-Delete header."""
        headers = {**self.headers, "X-Confirm-Delete": "true"}
        response = requests.delete(
            f"{self.base_url}/connections/{conn_id}", headers=headers
        )
        response.raise_for_status()


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

# Create a MySQL connection
result = api.create_connection({
    "name": "mysql-source",
    "type": "mysql",
    "spec": {
        "database": {
            "host": "localhost",
            "port": 3306,
            "username": "root",
            "password": "secret",
            "database": "source_db",
        }
    },
})
print(f"Created: {result['id']}")

# Test connectivity
ping = api.ping(result["id"])
print(f"Ping: {ping['ping']}")

# List databases and tables
databases = api.list_databases(result["id"])
for db in databases:
    if not db["isSystem"]:
        tables = api.list_tables(result["id"], db["name"])
        print(f"{db['name']}: {tables}")

import axios, { AxiosInstance } from 'axios'

interface ConnectionSpec {
  database?: {
    host: string
    port: number
    username: string
    password: string
    database?: string
  }
  snowflake?: {
    account: string
    username: string
    password: string
    database?: string
    warehouse?: string
  }
  s3?: {
    region: string
    credentials?: { accessKey: string; secretKey: string }
    endpoint?: string
  }
}

interface CreateConnectionRequest {
  name: string
  type: 'mysql' | 'postgresql' | 'snowflake' | 'files' | 's3' | 'gcs' | 'azure'
  spec: ConnectionSpec
}

interface ConnectionResult {
  id: string
  created: number
}

interface PingResponse {
  ping: 'ok' | 'failed'
  error?: string
}

interface ConnectionListResponse {
  items: Array<{
    id: string
    name: string
    type: string
    spec: ConnectionSpec
    created: number
  }>
  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 listConnections(): Promise<ConnectionListResponse> {
    const { data } = await this.client.get('/connections')
    return data
  }

  async createConnection(req: CreateConnectionRequest): Promise<ConnectionResult> {
    const { data } = await this.client.post('/connections', req)
    return data
  }

  async ping(connId: string): Promise<PingResponse> {
    const { data } = await this.client.post(`/connections/${connId}/ping`)
    return data
  }

  async listDatabases(connId: string): Promise<Array<{ name: string; isSystem: boolean }>> {
    const { data } = await this.client.get(`/connections/${connId}/databases`)
    return data
  }

  async listTables(connId: string, database: string): Promise<string[]> {
    const { data } = await this.client.get(
      `/connections/${connId}/databases/${database}/tables`
    )
    return data
  }

  async deleteConnection(connId: string): Promise<void> {
    await this.client.delete(`/connections/${connId}`, {
      headers: { 'X-Confirm-Delete': 'true' },
    })
  }
}

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

  const result = await api.createConnection({
    name: 'mysql-source',
    type: 'mysql',
    spec: {
      database: {
        host: 'localhost',
        port: 3306,
        username: 'root',
        password: 'secret',
        database: 'source_db',
      },
    },
  })
  console.log('Created:', result.id)

  const ping = await api.ping(result.id)
  console.log('Ping:', ping.ping)

  const databases = await api.listDatabases(result.id)
  for (const db of databases.filter((d) => !d.isSystem)) {
    const tables = await api.listTables(result.id, db.name)
    console.log(`${db.name}:`, tables)
  }
}

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] ListConnections() {
        return Invoke-RestMethod -Uri "$($this.BaseURL)/connections" `
            -Headers $this.Headers -Method Get
    }

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

    [object] Ping([string]$ConnId) {
        return Invoke-RestMethod -Uri "$($this.BaseURL)/connections/$ConnId/ping" `
            -Headers $this.Headers -Method Post
    }

    [object] ListDatabases([string]$ConnId) {
        return Invoke-RestMethod -Uri "$($this.BaseURL)/connections/$ConnId/databases" `
            -Headers $this.Headers -Method Get
    }

    [object] ListTables([string]$ConnId, [string]$Database) {
        return Invoke-RestMethod `
            -Uri "$($this.BaseURL)/connections/$ConnId/databases/$Database/tables" `
            -Headers $this.Headers -Method Get
    }

    [void] DeleteConnection([string]$ConnId) {
        $deleteHeaders = $this.Headers.Clone()
        $deleteHeaders["X-Confirm-Delete"] = "true"
        Invoke-RestMethod -Uri "$($this.BaseURL)/connections/$ConnId" `
            -Headers $deleteHeaders -Method Delete
    }
}

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

# Create a MySQL connection
$result = $api.CreateConnection(@{
    name = "mysql-source"
    type = "mysql"
    spec = @{
        database = @{
            host     = "localhost"
            port     = 3306
            username = "root"
            password = "secret"
            database = "source_db"
        }
    }
})
Write-Host "Created: $($result.id)"

# Test connectivity
$ping = $api.Ping($result.id)
Write-Host "Ping: $($ping.ping)"

# List databases and tables
$databases = $api.ListDatabases($result.id)
foreach ($db in $databases | Where-Object { -not $_.isSystem }) {
    $tables = $api.ListTables($result.id, $db.name)
    Write-Host "$($db.name): $($tables -join ', ')"
}