API

# Health check
curl <laminar_backend>/api/v1/ping
 
# List pipelines
curl <laminar_backend>/api/v1/pipelines
 
# With authentication (if enabled)
curl -H "Authorization: Bearer <token>" <laminar_backend>/api/v1/pipelines

curl -X POST <laminar_backend>/api/v1/pipelines \
  -H "Content-Type: application/json" \
  -d '{
    "name": "my-pipeline",
    "query": "INSERT INTO sink SELECT * FROM source",
    "parallelism": 4,
    "checkpoint_interval_micros": 10000000
  }'

{
  "id": "pl_abc123",
  "name": "my-pipeline",
  "query": "INSERT INTO sink SELECT * FROM source",
  "checkpoint_interval_micros": 10000000,
  "stop": "none",
  "createdAt": 1699900000000,
  "actionText": "Running",
  "actionInProgress": false,
  "preview": false,
  "graph": {
    "nodes": [...],
    "edges": [...]
  }
}

# Stop pipeline gracefully
curl -X PATCH <laminar_backend>/api/v1/pipelines/pl_abc123 \
  -H "Content-Type: application/json" \
  -d '{"stop": "graceful"}'
 
# Resume pipeline
curl -X PATCH <laminar_backend>/api/v1/pipelines/pl_abc123 \
  -H "Content-Type: application/json" \
  -d '{"stop": "none"}'
 
# Update parallelism
curl -X PATCH <laminar_backend>/api/v1/pipelines/pl_abc123 \
  -H "Content-Type: application/json" \
  -d '{"parallelism": 8}'

curl -X POST <laminar_backend>/api/v1/pipelines/validate_query \
  -H "Content-Type: application/json" \
  -d '{
    "query": "INSERT INTO sink SELECT * FROM source"
  }'

curl -N <laminar_backend>/api/v1/pipelines/pl_abc123/jobs/job_xyz/output

# No authentication
curl -X POST <laminar_backend>/api/v1/connectors/kafka/profiles \
  -H "Content-Type: application/json" \
  -d '{
    "name": "local-kafka",
    "config": {
      "bootstrap_servers": "localhost:9092",
      "authentication": {"type": "none"}
    }
  }'
 
# SASL authentication
curl -X POST <laminar_backend>/api/v1/connectors/kafka/profiles \
  -H "Content-Type: application/json" \
  -d '{
    "name": "prod-kafka",
    "config": {
      "bootstrap_servers": ["broker1:9092", "broker2:9092"],
      "authentication": {
        "type": "sasl",
        "sasl_config": {
          "mechanism": "PLAIN",
          "username": "user",
          "password": "pass",
          "protocol": "SASL_SSL"
        }
      },
      "schema_registry": {
        "endpoint": "https://schema-registry:8081",
        "api_key": "key",
        "api_secret": "secret"
      }
    }
  }'

# REST catalog
curl -X POST <laminar_backend>/api/v1/connectors/iceberg/profiles \
  -H "Content-Type: application/json" \
  -d '{
    "name": "iceberg-rest",
    "config": {
      "catalog": {
        "type": "rest",
        "rest_catalog_config": {
          "uri": "http://localhost:8181"
        }
      }
    }
  }'
 
# AWS Glue catalog
curl -X POST <laminar_backend>/api/v1/connectors/iceberg/profiles \
  -H "Content-Type: application/json" \
  -d '{
    "name": "iceberg-glue",
    "config": {
      "catalog": {
        "type": "glue",
        "glue_catalog_config": {
          "database_name": "my_database",
          "region": "us-east-1"
        }
      }
    }
  }'

# Kafka source table
curl -X POST <laminar_backend>/api/v1/connectors/kafka/tables \
  -H "Content-Type: application/json" \
  -d '{
    "name": "orders",
    "connection_profile_id": "prof_abc123",
    "config": {
      "topic": "orders",
      "type": {
        "source_config": {
          "offset": "latest"
        }
      }
    },
    "schema": {
      "format": {"json": {}},
      "fields": [
        {"field_name": "id", "field_type": {"primitive": "Int64"}, "nullable": false},
        {"field_name": "amount", "field_type": {"primitive": "Float64"}, "nullable": false},
        {"field_name": "customer", "field_type": {"primitive": "Utf8"}, "nullable": true}
      ]
    }
  }'
 
# Kafka sink table
curl -X POST <laminar_backend>/api/v1/connectors/kafka/tables \
  -H "Content-Type: application/json" \
  -d '{
    "name": "processed-orders",
    "connection_profile_id": "prof_abc123",
    "config": {
      "topic": "processed-orders",
      "type": {
        "sink_config": {}
      }
    },
    "schema": {
      "format": {"json": {}}
    }
  }'

curl -X POST <laminar_backend>/api/v1/connectors/iceberg/tables \
  -H "Content-Type: application/json" \
  -d '{
    "name": "events_iceberg",
    "connection_profile_id": "prof_iceberg123",
    "config": {
      "database": "analytics",
      "table": "events",
      "type": {
        "sink_config": {
          "commit_interval_micros": 60000000
        }
      }
    }
  }'

curl <laminar_backend>/api/v1/connectors

# First page
curl "<laminar_backend>/api/v1/pipelines?limit=10"
 
# Next page (using last item's ID)
curl "<laminar_backend>/api/v1/pipelines?limit=10&starting_after=pl_abc123"

{
  "data": [...],
  "has_more": true
}

{
  "error": "Pipeline not found: pl_abc123"
}

import requests
 
API_URL = "<laminar_backend>/api/v1"
 
# List pipelines
response = requests.get(f"{API_URL}/pipelines")
pipelines = response.json()["data"]
 
# Create pipeline
pipeline = requests.post(f"{API_URL}/pipelines", json={
    "name": "my-pipeline",
    "query": "INSERT INTO sink SELECT * FROM source",
    "parallelism": 4
}).json()
 
# Stop pipeline gracefully
requests.patch(f"{API_URL}/pipelines/{pipeline['id']}", json={
    "stop": "graceful"
})
 
# Get job status
jobs = requests.get(f"{API_URL}/pipelines/{pipeline['id']}/jobs").json()["data"]

const API_URL = "<laminar_backend>/api/v1";
 
// List pipelines
const pipelines = await fetch(`${API_URL}/pipelines`)
  .then(r => r.json())
  .then(d => d.data);
 
// Create pipeline
const pipeline = await fetch(`${API_URL}/pipelines`, {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({
    name: "my-pipeline",
    query: "INSERT INTO sink SELECT * FROM source",
    parallelism: 4
  })
}).then(r => r.json());
 
// Stop pipeline
await fetch(`${API_URL}/pipelines/${pipeline.id}`, {
  method: "PATCH",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ stop: "graceful" })
});

# 1. Create a Kafka profile
PROFILE=$(curl -s -X POST <laminar_backend>/api/v1/connectors/kafka/profiles \
  -H "Content-Type: application/json" \
  -d '{
    "name": "local-kafka",
    "config": {
      "bootstrap_servers": "localhost:9092",
      "authentication": {"type": "none"}
    }
  }')
PROFILE_ID=$(echo $PROFILE | jq -r '.id')
 
# 2. Create source table
curl -s -X POST <laminar_backend>/api/v1/connectors/kafka/tables \
  -H "Content-Type: application/json" \
  -d "{
    \"name\": \"orders\",
    \"connection_profile_id\": \"$PROFILE_ID\",
    \"config\": {
      \"topic\": \"orders\",
      \"type\": {\"source_config\": {\"offset\": \"latest\"}}
    },
    \"schema\": {
      \"format\": {\"json\": {}},
      \"fields\": [
        {\"field_name\": \"id\", \"field_type\": {\"primitive\": \"Int64\"}, \"nullable\": false},
        {\"field_name\": \"amount\", \"field_type\": {\"primitive\": \"Float64\"}, \"nullable\": false}
      ]
    }
  }"
 
# 3. Create sink table
curl -s -X POST <laminar_backend>/api/v1/connectors/kafka/tables \
  -H "Content-Type: application/json" \
  -d "{
    \"name\": \"large-orders\",
    \"connection_profile_id\": \"$PROFILE_ID\",
    \"config\": {
      \"topic\": \"large-orders\",
      \"type\": {\"sink_config\": {}}
    },
    \"schema\": {\"format\": {\"json\": {}}}
  }"
 
# 4. Create pipeline
PIPELINE=$(curl -s -X POST <laminar_backend>/api/v1/pipelines \
  -H "Content-Type: application/json" \
  -d '{
    "name": "filter-large-orders",
    "query": "INSERT INTO `large-orders` SELECT * FROM orders WHERE amount > 100",
    "parallelism": 4
  }')
PIPELINE_ID=$(echo $PIPELINE | jq -r '.id')
 
# 5. Check status
curl -s <laminar_backend>/api/v1/pipelines/$PIPELINE_ID | jq
 
# 6. List jobs
curl -s <laminar_backend>/api/v1/pipelines/$PIPELINE_ID/jobs | jq
 
# 7. Stop pipeline
curl -s -X PATCH <laminar_backend>/api/v1/pipelines/$PIPELINE_ID \
  -H "Content-Type: application/json" \
  -d '{"stop": "graceful"}'