curl --request GET \
--url https://api.streamkap.com/dashboard/lineage/topics \
--header 'Authorization: Bearer <token>'import requests
url = "https://api.streamkap.com/dashboard/lineage/topics"
headers = {"Authorization": "Bearer <token>"}
response = requests.get(url, headers=headers)
print(response.text)const options = {method: 'GET', headers: {Authorization: 'Bearer <token>'}};
fetch('https://api.streamkap.com/dashboard/lineage/topics', options)
.then(res => res.json())
.then(res => console.log(res))
.catch(err => console.error(err));<?php
$curl = curl_init();
curl_setopt_array($curl, [
CURLOPT_URL => "https://api.streamkap.com/dashboard/lineage/topics",
CURLOPT_RETURNTRANSFER => true,
CURLOPT_ENCODING => "",
CURLOPT_MAXREDIRS => 10,
CURLOPT_TIMEOUT => 30,
CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
CURLOPT_CUSTOMREQUEST => "GET",
CURLOPT_HTTPHEADER => [
"Authorization: Bearer <token>"
],
]);
$response = curl_exec($curl);
$err = curl_error($curl);
curl_close($curl);
if ($err) {
echo "cURL Error #:" . $err;
} else {
echo $response;
}package main
import (
"fmt"
"net/http"
"io"
)
func main() {
url := "https://api.streamkap.com/dashboard/lineage/topics"
req, _ := http.NewRequest("GET", url, nil)
req.Header.Add("Authorization", "Bearer <token>")
res, _ := http.DefaultClient.Do(req)
defer res.Body.Close()
body, _ := io.ReadAll(res.Body)
fmt.Println(string(body))
}HttpResponse<String> response = Unirest.get("https://api.streamkap.com/dashboard/lineage/topics")
.header("Authorization", "Bearer <token>")
.asString();require 'uri'
require 'net/http'
url = URI("https://api.streamkap.com/dashboard/lineage/topics")
http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true
request = Net::HTTP::Get.new(url)
request["Authorization"] = 'Bearer <token>'
response = http.request(request)
puts response.read_body{
"page": 1,
"page_size": 10,
"total": 123,
"has_next": true,
"result": [
{
"id": "<string>",
"name": "<string>",
"source": {
"id": "<string>",
"name": "<string>",
"connector": "",
"snapshot_progress": {
"state": "<string>",
"snapshot_type": "incremental",
"total_topic_count": 0,
"remaining_topic_count": 0,
"done_topic_count": 0,
"cancelled_topic_count": 0,
"failed_topic_count": 0,
"total_table_count": 0,
"remaining_table_count": 0,
"rows_total": 0,
"rows_scanned": 0,
"progress_type": "tables",
"percent_complete": 0,
"eta_ms": 0,
"start_timestamp": "<string>",
"end_timestamp": "<string>"
}
},
"metrics": {
"StreamingMilliSecondsBehindSource": 0,
"SnapshotPercentageComplete": 0,
"recordSendTotal": 0,
"sourceRecordWriteTotal": 0,
"snapshotPercentageComplete": 0,
"recordsLag": 0,
"byteTotal": 0,
"sourceRecordPollRate": 0,
"sourceQaCount": 123
},
"snapshot_status": "<string>",
"snapshot_details": [
{
"status": "<string>",
"start_timestamp": "<string>",
"end_timestamp": "<string>",
"duration_ms": 123,
"rows_scanned": 0,
"rows_total": 0,
"trace": "<string>",
"eta_ms": 0
}
],
"transforms": [
{
"id": "<string>",
"name": "<string>",
"status": "",
"metrics": {
"latency": 0
},
"output_topics": [
"<string>"
],
"transform_type": "fan_out",
"destinations": [
{
"id": "<string>",
"name": "<string>",
"connector": "",
"pipeline_id": "",
"pipeline_status": "",
"lag": 123,
"latency_ms": 123,
"qaCount": 123,
"e2e_eta_ms": 123
}
]
}
],
"destinations": [
{
"id": "<string>",
"name": "<string>",
"connector": "",
"pipeline_id": "",
"pipeline_status": "",
"lag": 123,
"latency_ms": 123,
"qaCount": 123,
"e2e_eta_ms": 123
}
]
}
]
}{
"detail": [
{
"loc": [
"<string>"
],
"msg": "<string>",
"type": "<string>",
"input": "<unknown>",
"ctx": {}
}
]
}Get dashboard lineage topics
Returns detailed topic-level lineage information including source, transforms, destinations, and per-topic metrics. Supports filtering and pagination.
curl --request GET \
--url https://api.streamkap.com/dashboard/lineage/topics \
--header 'Authorization: Bearer <token>'import requests
url = "https://api.streamkap.com/dashboard/lineage/topics"
headers = {"Authorization": "Bearer <token>"}
response = requests.get(url, headers=headers)
print(response.text)const options = {method: 'GET', headers: {Authorization: 'Bearer <token>'}};
fetch('https://api.streamkap.com/dashboard/lineage/topics', options)
.then(res => res.json())
.then(res => console.log(res))
.catch(err => console.error(err));<?php
$curl = curl_init();
curl_setopt_array($curl, [
CURLOPT_URL => "https://api.streamkap.com/dashboard/lineage/topics",
CURLOPT_RETURNTRANSFER => true,
CURLOPT_ENCODING => "",
CURLOPT_MAXREDIRS => 10,
CURLOPT_TIMEOUT => 30,
CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
CURLOPT_CUSTOMREQUEST => "GET",
CURLOPT_HTTPHEADER => [
"Authorization: Bearer <token>"
],
]);
$response = curl_exec($curl);
$err = curl_error($curl);
curl_close($curl);
if ($err) {
echo "cURL Error #:" . $err;
} else {
echo $response;
}package main
import (
"fmt"
"net/http"
"io"
)
func main() {
url := "https://api.streamkap.com/dashboard/lineage/topics"
req, _ := http.NewRequest("GET", url, nil)
req.Header.Add("Authorization", "Bearer <token>")
res, _ := http.DefaultClient.Do(req)
defer res.Body.Close()
body, _ := io.ReadAll(res.Body)
fmt.Println(string(body))
}HttpResponse<String> response = Unirest.get("https://api.streamkap.com/dashboard/lineage/topics")
.header("Authorization", "Bearer <token>")
.asString();require 'uri'
require 'net/http'
url = URI("https://api.streamkap.com/dashboard/lineage/topics")
http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true
request = Net::HTTP::Get.new(url)
request["Authorization"] = 'Bearer <token>'
response = http.request(request)
puts response.read_body{
"page": 1,
"page_size": 10,
"total": 123,
"has_next": true,
"result": [
{
"id": "<string>",
"name": "<string>",
"source": {
"id": "<string>",
"name": "<string>",
"connector": "",
"snapshot_progress": {
"state": "<string>",
"snapshot_type": "incremental",
"total_topic_count": 0,
"remaining_topic_count": 0,
"done_topic_count": 0,
"cancelled_topic_count": 0,
"failed_topic_count": 0,
"total_table_count": 0,
"remaining_table_count": 0,
"rows_total": 0,
"rows_scanned": 0,
"progress_type": "tables",
"percent_complete": 0,
"eta_ms": 0,
"start_timestamp": "<string>",
"end_timestamp": "<string>"
}
},
"metrics": {
"StreamingMilliSecondsBehindSource": 0,
"SnapshotPercentageComplete": 0,
"recordSendTotal": 0,
"sourceRecordWriteTotal": 0,
"snapshotPercentageComplete": 0,
"recordsLag": 0,
"byteTotal": 0,
"sourceRecordPollRate": 0,
"sourceQaCount": 123
},
"snapshot_status": "<string>",
"snapshot_details": [
{
"status": "<string>",
"start_timestamp": "<string>",
"end_timestamp": "<string>",
"duration_ms": 123,
"rows_scanned": 0,
"rows_total": 0,
"trace": "<string>",
"eta_ms": 0
}
],
"transforms": [
{
"id": "<string>",
"name": "<string>",
"status": "",
"metrics": {
"latency": 0
},
"output_topics": [
"<string>"
],
"transform_type": "fan_out",
"destinations": [
{
"id": "<string>",
"name": "<string>",
"connector": "",
"pipeline_id": "",
"pipeline_status": "",
"lag": 123,
"latency_ms": 123,
"qaCount": 123,
"e2e_eta_ms": 123
}
]
}
],
"destinations": [
{
"id": "<string>",
"name": "<string>",
"connector": "",
"pipeline_id": "",
"pipeline_status": "",
"lag": 123,
"latency_ms": 123,
"qaCount": 123,
"e2e_eta_ms": 123
}
]
}
]
}{
"detail": [
{
"loc": [
"<string>"
],
"msg": "<string>",
"type": "<string>",
"input": "<unknown>",
"ctx": {}
}
]
}Authorizations
Bearer authentication header of the form Bearer <token>, where <token> is your auth token.
Query Parameters
Filter by source entity ID
Filter by destination entity ID
Filter by transform entity ID
Filter by topic name (substring match)
Comma-separated topic filters: has_destination, orphaned, has_transform, dlq, snapshotting
When false, exclude DLQ topics
Comma-separated snapshot status filter: running, completed, pending
Page number (1-indexed)
x >= 1Results per page
1 <= x <= 100Sort field
byteTotal, latency, speed, lag, name Sort direction
asc, desc When false, skip ClickHouse metrics and return metrics as null for faster initial render
Response
Successful Response
Response model for GET /dashboard/lineage/topics.
Was this page helpful?