Streaming session service#

OpenAPI file: StreamingSessionService.json

Endpoints#

Resolve#

POST /ingest/v1/internal/streaming-session/dataset/{datasetRid}/resolve

Path parameters

datasetRid string (RID) required

​

Body

source StreamingSessionSource

The source attribution for this streaming session. Each unique source gets its own in-progress session per dataset. If absent, the session is unattributed.

Set type to "mesh".

mesh MeshStreamingSessionSource

​

Fields of MeshStreamingSessionSource
sourceDatasetRid string (RID) required

​

sourceOrgRid string (RID) required

​

originStreamingSessionSource string required

​

Set type to "dataConnector".

dataConnector DataConnectorStreamingSessionSource

​

Fields of DataConnectorStreamingSessionSource
dataConnectorRid string (RID) required

​

Set type to "custom".

custom CustomStreamingSessionSource

​

Fields of CustomStreamingSessionSource
sourceName string required

​

Response

sessionRid string (RID)

​

datasetRid string (RID)

​

status StreamingSessionStatus

​

Values of StreamingSessionStatus
IN_PROGRESS

​

COMPLETED

​

source StreamingSessionSource

​

Set type to "mesh".

mesh MeshStreamingSessionSource

​

Fields of MeshStreamingSessionSource
sourceDatasetRid string (RID)

​

sourceOrgRid string (RID)

​

originStreamingSessionSource string

​

Set type to "dataConnector".

dataConnector DataConnectorStreamingSessionSource

​

Fields of DataConnectorStreamingSessionSource
dataConnectorRid string (RID)

​

Set type to "custom".

custom CustomStreamingSessionSource

​

Fields of CustomStreamingSessionSource
sourceName string

​

bounds Range

​

Fields of Range
start Timestamp

​

Fields of Timestamp
seconds integer (int64)

​

nanos integer (int64)

​

end Timestamp

​

pointsCount integer (int64)

​

createdAt Timestamp

​

Request

curl -X POST "https://api.gov.nominal.io/api/ingest/v1/internal/streaming-session/dataset/<dataset rid>/resolve" \
  -H "Authorization: Bearer <api key>" \
  -H "Content-Type: application/json" \
  --data '{
  "source": {
    "type": "mesh",
    "mesh": {
      "sourceDatasetRid": "<dataset rid>",
      "sourceOrgRid": "<org rid>",
      "originStreamingSessionSource": "<origin streaming session source>"
    }
  }
}'

import requests

response = requests.post(
    "https://api.gov.nominal.io/api/ingest/v1/internal/streaming-session/dataset/<dataset rid>/resolve",
    headers={"Authorization": "Bearer <api key>"},
    json={
        "source": {
            "type": "mesh",
            "mesh": {
                "sourceDatasetRid": "<dataset rid>",
                "sourceOrgRid": "<org rid>",
                "originStreamingSessionSource": "<origin streaming session source>",
            },
        },
    },
)
response.raise_for_status()
print(response.json())

import httpx

response = httpx.post(
    "https://api.gov.nominal.io/api/ingest/v1/internal/streaming-session/dataset/<dataset rid>/resolve",
    headers={"Authorization": "Bearer <api key>"},
    json={
        "source": {
            "type": "mesh",
            "mesh": {
                "sourceDatasetRid": "<dataset rid>",
                "sourceOrgRid": "<org rid>",
                "originStreamingSessionSource": "<origin streaming session source>",
            },
        },
    },
)
response.raise_for_status()
print(response.json())

import asyncio

import aiohttp


async def main() -> None:
    async with aiohttp.ClientSession() as session:
        async with session.post(
            "https://api.gov.nominal.io/api/ingest/v1/internal/streaming-session/dataset/<dataset rid>/resolve",
            headers={"Authorization": "Bearer <api key>"},
            json={
                "source": {
                    "type": "mesh",
                    "mesh": {
                        "sourceDatasetRid": "<dataset rid>",
                        "sourceOrgRid": "<org rid>",
                        "originStreamingSessionSource": "<origin streaming session source>",
                    },
                },
            },
        ) as response:
            response.raise_for_status()
            print(await response.json())


asyncio.run(main())

Response: 200

{
  "sessionRid": "<streaming session rid>",
  "datasetRid": "<dataset rid>",
  "status": "IN_PROGRESS",
  "source": {
    "type": "mesh",
    "mesh": {
      "sourceDatasetRid": "<dataset rid>",
      "sourceOrgRid": "<org rid>",
      "originStreamingSessionSource": "<origin streaming session source>"
    }
  },
  "bounds": {
    "start": {
      "seconds": 1,
      "nanos": 1
    },
    "end": {
      "seconds": 1,
      "nanos": 1
    }
  },
  "pointsCount": 1,
  "createdAt": {
    "seconds": 1,
    "nanos": 1
  }
}

Heartbeat#

PUT /ingest/v1/internal/streaming-session/{sessionRid}/heartbeat

Path parameters

sessionRid string (RID) required

​

Body

bounds Range required

The time range of data points ingested since the last heartbeat. The server atomically merges this with the session’s existing bounds using LEAST/GREATEST to track the full time range.

Fields of Range
start Timestamp required

​

Fields of Timestamp
seconds integer (int64) required

​

nanos integer (int64) required

​

end Timestamp required

​

pointsCount integer required

Number of data points ingested since the last heartbeat. The server atomically adds this to the session’s running total.

Response

No content (HTTP 204).

Request

curl -X PUT "https://api.gov.nominal.io/api/ingest/v1/internal/streaming-session/<session rid>/heartbeat" \
  -H "Authorization: Bearer <api key>" \
  -H "Content-Type: application/json" \
  --data '{
  "bounds": {
    "start": {
      "seconds": 1,
      "nanos": 1
    },
    "end": {
      "seconds": 1,
      "nanos": 1
    }
  },
  "pointsCount": 1
}'

import requests

response = requests.put(
    "https://api.gov.nominal.io/api/ingest/v1/internal/streaming-session/<session rid>/heartbeat",
    headers={"Authorization": "Bearer <api key>"},
    json={
        "bounds": {
            "start": {"seconds": 1, "nanos": 1},
            "end": {"seconds": 1, "nanos": 1},
        },
        "pointsCount": 1,
    },
)
response.raise_for_status()

import httpx

response = httpx.put(
    "https://api.gov.nominal.io/api/ingest/v1/internal/streaming-session/<session rid>/heartbeat",
    headers={"Authorization": "Bearer <api key>"},
    json={
        "bounds": {
            "start": {"seconds": 1, "nanos": 1},
            "end": {"seconds": 1, "nanos": 1},
        },
        "pointsCount": 1,
    },
)
response.raise_for_status()

import asyncio

import aiohttp


async def main() -> None:
    async with aiohttp.ClientSession() as session:
        async with session.put(
            "https://api.gov.nominal.io/api/ingest/v1/internal/streaming-session/<session rid>/heartbeat",
            headers={"Authorization": "Bearer <api key>"},
            json={
                "bounds": {
                    "start": {"seconds": 1, "nanos": 1},
                    "end": {"seconds": 1, "nanos": 1},
                },
                "pointsCount": 1,
            },
        ) as response:
            response.raise_for_status()


asyncio.run(main())

Response: 204

No content.

Search streaming sessions#

POST /ingest/v1/streaming-sessions/search

Returns a paginated list of streaming sessions, optionally filtered by dataset.

Body

pageSize integer

Defaults to 100. Clamped to 1_000 if larger, and to 1 if non-positive.

nextPageToken string

​

filter StreamingSessionSearchFilter

​

Set type to "datasetRids".

datasetRids array of string (RID)

​

Set type to "source".

source StreamingSourceFilter

​

Fields of StreamingSourceFilter
datasetRid string (RID) required

​

source StreamingSessionSource

​

Set type to "mesh".

mesh MeshStreamingSessionSource

​

Fields of MeshStreamingSessionSource
sourceDatasetRid string (RID) required

​

sourceOrgRid string (RID) required

​

originStreamingSessionSource string required

​

Set type to "dataConnector".

dataConnector DataConnectorStreamingSessionSource

​

Fields of DataConnectorStreamingSessionSource
dataConnectorRid string (RID) required

​

Set type to "custom".

custom CustomStreamingSessionSource

​

Fields of CustomStreamingSessionSource
sourceName string required

​

Set type to "timeRange".

timeRange StreamingSessionTimeRangeFilter

​

Fields of StreamingSessionTimeRangeFilter
start Timestamp

​

Fields of Timestamp
seconds integer (int64) required

​

nanos integer (int64) required

​

end Timestamp

​

Set type to "workspace".

workspace string (RID)

Matches sessions whose dataset belongs to the workspace.

Set type to "and".

and array of StreamingSessionSearchFilter

​

Set type to "or".

or array of StreamingSessionSearchFilter

​

Set type to "not".

not StreamingSessionSearchFilter

​

sort StreamingSessionSortOptions

Defaults to created_at DESC when omitted.

Fields of StreamingSessionSortOptions
isDescending boolean required

​

sortKey StreamingSessionSortKey required

​

Values of StreamingSessionSortKey
CREATED_AT

Orders by the time the session was created.

DATA_BOUNDS_START

Orders by the start of the session’s data bounds. Sessions with no data bounds are excluded from the results.

Response

streamingSessions array of StreamingSession

​

Fields of StreamingSession
sessionRid string (RID)

​

datasetRid string (RID)

​

status StreamingSessionStatus

​

Values of StreamingSessionStatus
IN_PROGRESS

​

COMPLETED

​

source StreamingSessionSource

​

Set type to "mesh".

mesh MeshStreamingSessionSource

​

Fields of MeshStreamingSessionSource
sourceDatasetRid string (RID)

​

sourceOrgRid string (RID)

​

originStreamingSessionSource string

​

Set type to "dataConnector".

dataConnector DataConnectorStreamingSessionSource

​

Fields of DataConnectorStreamingSessionSource
dataConnectorRid string (RID)

​

Set type to "custom".

custom CustomStreamingSessionSource

​

Fields of CustomStreamingSessionSource
sourceName string

​

bounds Range

​

Fields of Range
start Timestamp

​

Fields of Timestamp
seconds integer (int64)

​

nanos integer (int64)

​

end Timestamp

​

pointsCount integer (int64)

​

createdAt Timestamp

​

lastSeenAt Timestamp

Time of the most recent heartbeat for this session.

closedAt Timestamp

Time the session was completed. Absent while the session is in progress.

nextPageToken string

​

Request

curl -X POST "https://api.gov.nominal.io/api/ingest/v1/streaming-sessions/search" \
  -H "Authorization: Bearer <api key>" \
  -H "Content-Type: application/json" \
  --data '{
  "pageSize": 1,
  "nextPageToken": "<token>",
  "filter": {
    "type": "datasetRids",
    "datasetRids": [
      "<dataset rid>"
    ]
  },
  "sort": {
    "isDescending": true,
    "sortKey": "CREATED_AT"
  }
}'

import requests

response = requests.post(
    "https://api.gov.nominal.io/api/ingest/v1/streaming-sessions/search",
    headers={"Authorization": "Bearer <api key>"},
    json={
        "pageSize": 1,
        "nextPageToken": "<token>",
        "filter": {"type": "datasetRids", "datasetRids": ["<dataset rid>"]},
        "sort": {"isDescending": True, "sortKey": "CREATED_AT"},
    },
)
response.raise_for_status()
print(response.json())

import httpx

response = httpx.post(
    "https://api.gov.nominal.io/api/ingest/v1/streaming-sessions/search",
    headers={"Authorization": "Bearer <api key>"},
    json={
        "pageSize": 1,
        "nextPageToken": "<token>",
        "filter": {"type": "datasetRids", "datasetRids": ["<dataset rid>"]},
        "sort": {"isDescending": True, "sortKey": "CREATED_AT"},
    },
)
response.raise_for_status()
print(response.json())

import asyncio

import aiohttp


async def main() -> None:
    async with aiohttp.ClientSession() as session:
        async with session.post(
            "https://api.gov.nominal.io/api/ingest/v1/streaming-sessions/search",
            headers={"Authorization": "Bearer <api key>"},
            json={
                "pageSize": 1,
                "nextPageToken": "<token>",
                "filter": {"type": "datasetRids", "datasetRids": ["<dataset rid>"]},
                "sort": {"isDescending": True, "sortKey": "CREATED_AT"},
            },
        ) as response:
            response.raise_for_status()
            print(await response.json())


asyncio.run(main())

Response: 200

{
  "streamingSessions": [
    {
      "sessionRid": "<streaming session rid>",
      "datasetRid": "<dataset rid>",
      "status": "IN_PROGRESS",
      "source": {
        "type": "mesh",
        "mesh": {
          "sourceDatasetRid": "<dataset rid>",
          "sourceOrgRid": "<org rid>",
          "originStreamingSessionSource": "<origin streaming session source>"
        }
      },
      "bounds": {
        "start": {
          "seconds": 1,
          "nanos": 1
        },
        "end": {
          "seconds": 1,
          "nanos": 1
        }
      },
      "pointsCount": 1,
      "createdAt": {
        "seconds": 1,
        "nanos": 1
      },
      "lastSeenAt": {
        "seconds": 1,
        "nanos": 1
      },
      "closedAt": {
        "seconds": 1,
        "nanos": 1
      }
    }
  ],
  "nextPageToken": "<token>"
}

Search streaming sources#

POST /ingest/v1/streaming-sources/search

Returns a paginated list of streaming sessions aggregated by (source, dataset) group, ordered by latest activity descending.

Body

pageSize integer

Defaults to 100. Clamped to 1_000 if larger, and to 1 if non-positive.

nextPageToken string

​

filter StreamingSourceSearchFilter

​

Set type to "datasetRids".

datasetRids array of string (RID)

​

Set type to "source".

source StreamingSourceFilter

​

Fields of StreamingSourceFilter
datasetRid string (RID) required

​

source StreamingSessionSource

​

Set type to "mesh".

mesh MeshStreamingSessionSource

​

Fields of MeshStreamingSessionSource
sourceDatasetRid string (RID) required

​

sourceOrgRid string (RID) required

​

originStreamingSessionSource string required

​

Set type to "dataConnector".

dataConnector DataConnectorStreamingSessionSource

​

Fields of DataConnectorStreamingSessionSource
dataConnectorRid string (RID) required

​

Set type to "custom".

custom CustomStreamingSessionSource

​

Fields of CustomStreamingSessionSource
sourceName string required

​

Set type to "workspace".

workspace string (RID)

Matches sources whose dataset belongs to the workspace.

Set type to "and".

and array of StreamingSourceSearchFilter

​

Set type to "or".

or array of StreamingSourceSearchFilter

​

Set type to "not".

not StreamingSourceSearchFilter

​

activityAfter Timestamp

If present, only sessions last seen strictly after this instant are aggregated; groups whose sessions were all last seen at or before it are omitted.

Fields of Timestamp
seconds integer (int64) required

​

nanos integer (int64) required

​

Response

streamingSources array of StreamingSource

​

Fields of StreamingSource
source StreamingSessionSource

The source attribution shared by the aggregated sessions. Absent for unattributed sessions.

Set type to "mesh".

mesh MeshStreamingSessionSource

​

Fields of MeshStreamingSessionSource
sourceDatasetRid string (RID)

​

sourceOrgRid string (RID)

​

originStreamingSessionSource string

​

Set type to "dataConnector".

dataConnector DataConnectorStreamingSessionSource

​

Fields of DataConnectorStreamingSessionSource
dataConnectorRid string (RID)

​

Set type to "custom".

custom CustomStreamingSessionSource

​

Fields of CustomStreamingSessionSource
sourceName string

​

datasetRid string (RID)

​

sessionCount integer (int64)

Number of sessions aggregated into this group.

totalPointsCount integer (int64)

Sum of the points counts across the aggregated sessions.

earliestCreatedAt Timestamp

Creation time of the oldest aggregated session.

Fields of Timestamp
seconds integer (int64)

​

nanos integer (int64)

​

latestLastSeenAt Timestamp

Most recent heartbeat time across the aggregated sessions.

mergedBounds Range

Union of the data bounds across the aggregated sessions, if any session has bounds.

Fields of Range
start Timestamp

​

end Timestamp

​

anyInProgress boolean

True if at least one aggregated session is in progress.

nextPageToken string

​

Request

curl -X POST "https://api.gov.nominal.io/api/ingest/v1/streaming-sources/search" \
  -H "Authorization: Bearer <api key>" \
  -H "Content-Type: application/json" \
  --data '{
  "pageSize": 1,
  "nextPageToken": "<token>",
  "filter": {
    "type": "datasetRids",
    "datasetRids": [
      "<dataset rid>"
    ]
  },
  "activityAfter": {
    "seconds": 1,
    "nanos": 1
  }
}'

import requests

response = requests.post(
    "https://api.gov.nominal.io/api/ingest/v1/streaming-sources/search",
    headers={"Authorization": "Bearer <api key>"},
    json={
        "pageSize": 1,
        "nextPageToken": "<token>",
        "filter": {"type": "datasetRids", "datasetRids": ["<dataset rid>"]},
        "activityAfter": {"seconds": 1, "nanos": 1},
    },
)
response.raise_for_status()
print(response.json())

import httpx

response = httpx.post(
    "https://api.gov.nominal.io/api/ingest/v1/streaming-sources/search",
    headers={"Authorization": "Bearer <api key>"},
    json={
        "pageSize": 1,
        "nextPageToken": "<token>",
        "filter": {"type": "datasetRids", "datasetRids": ["<dataset rid>"]},
        "activityAfter": {"seconds": 1, "nanos": 1},
    },
)
response.raise_for_status()
print(response.json())

import asyncio

import aiohttp


async def main() -> None:
    async with aiohttp.ClientSession() as session:
        async with session.post(
            "https://api.gov.nominal.io/api/ingest/v1/streaming-sources/search",
            headers={"Authorization": "Bearer <api key>"},
            json={
                "pageSize": 1,
                "nextPageToken": "<token>",
                "filter": {"type": "datasetRids", "datasetRids": ["<dataset rid>"]},
                "activityAfter": {"seconds": 1, "nanos": 1},
            },
        ) as response:
            response.raise_for_status()
            print(await response.json())


asyncio.run(main())

Response: 200

{
  "streamingSources": [
    {
      "source": {
        "type": "mesh",
        "mesh": {
          "sourceDatasetRid": "<dataset rid>",
          "sourceOrgRid": "<org rid>",
          "originStreamingSessionSource": "<origin streaming session source>"
        }
      },
      "datasetRid": "<dataset rid>",
      "sessionCount": 1,
      "totalPointsCount": 1,
      "earliestCreatedAt": {
        "seconds": 1,
        "nanos": 1
      },
      "latestLastSeenAt": {
        "seconds": 1,
        "nanos": 1
      },
      "mergedBounds": {
        "start": {
          "seconds": 1,
          "nanos": 1
        },
        "end": {
          "seconds": 1,
          "nanos": 1
        }
      },
      "anyInProgress": true
    }
  ],
  "nextPageToken": "<token>"
}

Types#

CustomStreamingSessionSource#

sourceName string required

​

DataConnectorStreamingSessionSource#

dataConnectorRid string (RID) required

​

MeshStreamingSessionSource#

sourceDatasetRid string (RID) required

​

sourceOrgRid string (RID) required

​

originStreamingSessionSource string required

​

Range#

Deprecated for new APIs. Define an API-owned range with the endpoint semantics it needs instead.

start Timestamp required

​

end Timestamp required

​

StreamingSession#

Public view of a streaming ingest session.

sessionRid string (RID) required

​

datasetRid string (RID) required

​

status StreamingSessionStatus required

​

source StreamingSessionSource

​

bounds Range

​

pointsCount integer (int64) required

​

createdAt Timestamp required

​

lastSeenAt Timestamp required

Time of the most recent heartbeat for this session.

closedAt Timestamp

Time the session was completed. Absent while the session is in progress.

StreamingSessionSearchFilter#

Filter for streaming-session search. An empty datasetRids list matches all sessions. Filters compose with and/or/not.

Set type to the name of one of the other fields, and set only that field.

datasetRids array of string (RID)

​

source StreamingSourceFilter

​

timeRange StreamingSessionTimeRangeFilter

​

workspace string (RID)

Matches sessions whose dataset belongs to the workspace.

and array of StreamingSessionSearchFilter

​

or array of StreamingSessionSearchFilter

​

not StreamingSessionSearchFilter

​

StreamingSessionSortKey#

CREATED_AT

Orders by the time the session was created.

DATA_BOUNDS_START

Orders by the start of the session’s data bounds. Sessions with no data bounds are excluded from the results.

StreamingSessionSortOptions#

isDescending boolean required

​

sortKey StreamingSessionSortKey required

​

StreamingSessionSource#

Set type to the name of one of the other fields, and set only that field.

mesh MeshStreamingSessionSource

​

dataConnector DataConnectorStreamingSessionSource

​

custom CustomStreamingSessionSource

​

StreamingSessionStatus#

IN_PROGRESS

​

COMPLETED

​

StreamingSessionTimeRangeFilter#

Matches sessions whose data bounds overlap this window. Sessions with no data bounds do not match. If both bounds are present, start must not be after end.

start Timestamp

​

end Timestamp

​

StreamingSource#

Aggregate view of the streaming sessions sharing a single (source, dataset) pair.

source StreamingSessionSource

The source attribution shared by the aggregated sessions. Absent for unattributed sessions.

datasetRid string (RID) required

​

sessionCount integer (int64) required

Number of sessions aggregated into this group.

totalPointsCount integer (int64) required

Sum of the points counts across the aggregated sessions.

earliestCreatedAt Timestamp required

Creation time of the oldest aggregated session.

latestLastSeenAt Timestamp required

Most recent heartbeat time across the aggregated sessions.

mergedBounds Range

Union of the data bounds across the aggregated sessions, if any session has bounds.

anyInProgress boolean required

True if at least one aggregated session is in progress.

StreamingSourceFilter#

Matches the sessions of a single streaming source: the sessions on datasetRid with exactly the given source attribution. An absent source matches unattributed sessions.

datasetRid string (RID) required

​

source StreamingSessionSource

​

StreamingSourceSearchFilter#

Filter for streaming-source search. An empty datasetRids list matches all sources. Filters compose with and/or/not.

Set type to the name of one of the other fields, and set only that field.

datasetRids array of string (RID)

​

source StreamingSourceFilter

​

workspace string (RID)

Matches sources whose dataset belongs to the workspace.

and array of StreamingSourceSearchFilter

​

or array of StreamingSourceSearchFilter

​

not StreamingSourceSearchFilter

​

Timestamp#

Deprecated for new APIs. Use google.protobuf.Timestamp for absolute instants in proto APIs.

This legacy nanosecond precision timestamp type is represented by an epoch time in seconds and a nanosecond offset. The nanosecond offset is from the start of the epoch second, so must be less than 1 billion.

seconds integer (int64) required

​

nanos integer (int64) required

​

Page design based on Scalar, MIT License.