Domo

Technical Preview

Overview

A client-mode connector for the Domo Platform APIs (DataSet, Simple, Stream, User, Group, Page, Activity Log and Embed APIs). Use it to land plant data in Domo, such as OEE per minute, counters at shift end or historian tags, and to bring Domo results back to the shop floor: run SQL on a DataSet and publish the rows for an andon board or HMI, or issue an embed token to show a Domo card on a panel. The connector DomoRestApi in DomoRestApi.lot has 33 events (20 GET, 6 POST, 6 PUT, 1 PATCH). 11 are writes: DatasetCreate, DatasetUpdate, DatasetImport, SimpleDatasetCreate, SimpleImport, StreamCreate, StreamUpdate, StreamExecutionCreate, StreamPartUpload, StreamExecutionCommit, StreamExecutionAbort. There are no DELETE events. Domo Platform APIs run on one host for every customer: set BASE_ADDRESS to https://api.domo.com (already set in the template), with no trailing slash and no /v1. Topics use analytics/domo/<object>/<verb>; replies land on <topic>/response. See the base guide: REST API Connectors.

Setup

Sign in at developer.domo.com with a Domo account of your instance and open My Account > New Client.
Give the client a name and the scopes the broker needs: data for DataSets and Streams, user for users and groups, dashboard for pages, audit for the activity log. Save the Client ID and Client Secret.
The client acts as the Domo user that created it. To write to a DataSet that user needs the Manage DataSet grant (or ownership of the DataSet).
Keep DOMO_CLIENT_ID, DOMO_CLIENT_SECRET and the derived DOMO_BASIC_AUTH (see below) in broker secrets and inject them with GET SECRET.

Authentication

OAuthToken calls GET /oauth/token?grant_type=client_credentials&scope=<scope> with Authorization: Basic <credentials>. basic_auth is the Base64 of <DOMO_CLIENT_ID>:<DOMO_CLIENT_SECRET>; store it as the secret DOMO_BASIC_AUTH. scope is a space-separated subset of the client’s scopes, URL-encoded (data, data%20user). The reply carries an access_token that is valid for about one hour. Every other event sends Authorization: Bearer <access_token> from the access_token field of the payload.

{"basic_auth": "<DOMO_BASIC_AUTH>", "scope": "data%20user"}

basic_auth and access_token grant API access as the client’s user. Never retain topics that carry them, restrict analytics/domo/auth/# and analytics/domo/embed/# with ACLs, and create a separate client for the broker. A wrong client ID or secret returns 401 with {"error":"invalid_client"}; an expired or malformed token returns 401 with a WWW-Authenticate: Bearer error="invalid_token" header. Ask for a new token before the hour runs out.

Events

Event	Request	Source topic	Notes
OAuthToken	`GET /oauth/token?grant_type=client_credentials&scope={value.json.scope}`	`analytics/domo/auth/token`	Get an access token (OAuth 2.0 client credentials, expires_in about 3600 s).
DatasetsList	`GET /v1/datasets?{value.json.query}`	`analytics/domo/datasets/list`	List DataSets (scope data).
DatasetGet	`GET /v1/datasets/{value.json.dataset_id}`	`analytics/domo/datasets/get`	DataSet details: schema, rows, columns, owner, updatedAt.
DatasetCreate	`POST /v1/datasets`	`analytics/domo/datasets/create`	Write. Create a DataSet with a schema (column types STRING, DECIMAL, LONG, DOUBLE, DATE, DATETIME).
DatasetUpdate	`PUT /v1/datasets/{value.json.dataset_id}`	`analytics/domo/datasets/update`	Write. Update DataSet metadata (name, description, pdpEnabled, schema).
DatasetImport	`PUT /v1/datasets/{value.json.dataset_id}/data?updateMethod={value.json.update_method}`	`analytics/domo/datasets/import`	Write. Import CSV rows (catalog event). update_method APPEND or REPLACE (uppercase; REPLACE replaces all rows).
DatasetExport	`GET /v1/datasets/{value.json.dataset_id}/data?{value.json.query}`	`analytics/domo/datasets/export`	Export DataSet rows as CSV (Accept text/csv).
DatasetQuery	`POST /v1/datasets/query/execute/{value.json.dataset_id}`	`analytics/domo/datasets/query`	Run SQL on one DataSet and publish columns, metadata and rows (read only; the table is always called table).
PdpPoliciesList	`GET /v1/datasets/{value.json.dataset_id}/policies`	`analytics/domo/datasets/policies/list`	Personalized Data Permission (PDP) policies of a DataSet (row-level filters).
PdpPolicyGet	`GET /v1/datasets/{value.json.dataset_id}/policies/{value.json.pdp_id}`	`analytics/domo/datasets/policies/get`	One PDP policy (filters, users, groups).
SimpleDatasetCreate	`POST /v1/json?name={value.json.name}&description={value.json.description}`	`analytics/domo/simple/create`	Write. Simple API: create a DataSet whose schema comes from the first JSON import. name / description URL-encoded (a space is %20).
SimpleImport	`PUT /v1/json/{value.json.dataset_id}/data?updateMethod={value.json.update_method}`	`analytics/domo/simple/import`	Write. Simple API: import a JSON array of row objects (Domo converts it to CSV). update_method APPEND or REPLACE.
StreamsList	`GET /v1/streams?{value.json.query}`	`analytics/domo/streams/list`	List Streams (bulk upload pipes).
StreamsSearch	`GET /v1/streams/search?q={value.json.q}&fields={value.json.fields}`	`analytics/domo/streams/search`	Find the Stream of a DataSet. q: dataSource.id:<dataset_id> or dataSource.owner.id:<user_id>. fields: all, or a comma list of id,dataset,updateMethod,createdAt,modifiedAt.
StreamGet	`GET /v1/streams/{value.json.stream_id}?{value.json.query}`	`analytics/domo/streams/get`	One Stream with its DataSet.
StreamCreate	`POST /v1/streams`	`analytics/domo/streams/create`	Write. Create a Stream and its DataSet (Stream imports only work on DataSets created this way). updateMethod APPEND or REPLACE; keyColumnNames + APPEND makes an upsert DataSet.
StreamUpdate	`PATCH /v1/streams/{value.json.stream_id}?updateMethod={value.json.update_method}`	`analytics/domo/streams/update`	Write. Change the Stream’s import behaviour (updateMethod is required as a query parameter). update_method APPEND or REPLACE.
StreamExecutionsList	`GET /v1/streams/{value.json.stream_id}/executions?{value.json.query}`	`analytics/domo/streams/executions/list`	Executions (upload runs) of a Stream.
StreamExecutionGet	`GET /v1/streams/{value.json.stream_id}/executions/{value.json.execution_id}`	`analytics/domo/streams/executions/get`	One execution: currentState ACTIVE, SUCCESS, ERROR or ABORTED.
StreamExecutionCreate	`POST /v1/streams/{value.json.stream_id}/executions`	`analytics/domo/streams/executions/create`	Write. Start an upload run (catalog event). Returns the execution id used by StreamPartUpload, StreamExecutionCommit and StreamExecutionAbort.
StreamPartUpload	`PUT /v1/streams/{value.json.stream_id}/executions/{value.json.execution_id}/part/{value.json.part_id}`	`analytics/domo/streams/executions/part`	Write. Upload one part of CSV rows (no header) to an active execution. part_id 1, 2, 3 … in increasing order; retry a failed part with the same id.
StreamExecutionCommit	`PUT /v1/streams/{value.json.stream_id}/executions/{value.json.execution_id}/commit`	`analytics/domo/streams/executions/commit`	Write. Commit the execution: Domo imports all parts into the DataSet (catalog event).
StreamExecutionAbort	`PUT /v1/streams/{value.json.stream_id}/executions/{value.json.execution_id}/abort`	`analytics/domo/streams/executions/abort`	Write. Abort an active execution (its parts are discarded; the DataSet is unchanged).
UsersList	`GET /v1/users?{value.json.query}`	`analytics/domo/users/list`	List users (scope user).
UserGet	`GET /v1/users/{value.json.user_id}`	`analytics/domo/users/get`	One user (name, email, role, groups).
GroupsList	`GET /v1/groups?{value.json.query}`	`analytics/domo/groups/list`	List groups (scope user).
GroupGet	`GET /v1/groups/{value.json.group_id}`	`analytics/domo/groups/get`	One group.
GroupUsersList	`GET /v1/groups/{value.json.group_id}/users?{value.json.query}`	`analytics/domo/groups/users`	User ids in a group.
PagesList	`GET /v1/pages?{value.json.query}`	`analytics/domo/pages/list`	List pages (dashboards, scope dashboard).
PageGet	`GET /v1/pages/{value.json.page_id}`	`analytics/domo/pages/get`	One page with its card ids, owners and visibility.
PageCollectionsGet	`GET /v1/pages/{value.json.page_id}/collections`	`analytics/domo/pages/collections`	Card collections on a page.
AuditLogList	`GET /v1/audit?start={value.json.start}&{value.json.query}`	`analytics/domo/audit/list`	Activity log entries (scope audit). start: epoch milliseconds (required).
CardEmbedToken	`POST /v1/cards/embed/auth`	`analytics/domo/embed/cards/token`	Issue an embed token for embedded cards (Domo Everywhere programmatic filtering) to show a Domo card in an HMI.

Payloads

Field	Used by	Content
`basic_auth`	`OAuthToken`	Base64 of `client_id:client_secret`
`scope`	`OAuthToken`	URL-encoded, space-separated scopes, for example `data%20user`
`access_token`	every event except `OAuthToken`	Token from `OAuthToken`
`dataset_id`, `stream_id`, `execution_id`, `part_id`, `pdp_id`, `user_id`, `group_id`, `page_id`	events with ids in the path	Domo ids (DataSet ids are UUIDs, the others are numbers)
`query`	list and export events, `StreamGet`, `AuditLogList`	URL query string without `?`, or `""`, for example `limit=50&offset=0&nameLike=OEE`
`update_method`	`DatasetImport`, `SimpleImport`, `StreamUpdate`	`APPEND` or `REPLACE`, uppercase (Domo falls back to `REPLACE` on anything else)
`csv`	`DatasetImport`, `StreamPartUpload`	CSV rows without a header, matching the DataSet schema, one row per line
`body`	JSON events	JSON object, or a JSON array of row objects for `SimpleImport`
`q`, `fields`	`StreamsSearch`	`dataSource.id:<dataset_id>` and `all`
`start`	`AuditLogList`	Epoch milliseconds (required)
`name`, `description`	`SimpleDatasetCreate`	URL-encoded DataSet name and description

Basic Syntax

An excerpt of the template (the full file has all 33 events and comments):

DEFINE ROUTE DomoRestApi WITH TYPE REST_API
    ADD REST_API_CONFIG
        WITH BASE_ADDRESS "https://api.domo.com"
        WITH ENABLE_CLIENT "true"
        WITH USE_SSL "true"
        WITH REQUEST_TIMEOUT '120'
        WITH DEFAULT_CONTENT_TYPE "application/json"
    ADD EVENT OAuthToken
        WITH SOURCE_TOPIC "analytics/domo/auth/token"
        WITH DESTINATION_TOPIC "analytics/domo/auth/token/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/oauth/token?grant_type=client_credentials&scope={value.json.scope}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "User-Agent" "Coreflux-Broker REST_API route (https://www.coreflux.org)"
            ADD "Authorization" "Basic {value.json.basic_auth}"
    ADD EVENT DatasetImport
        WITH SOURCE_TOPIC "analytics/domo/datasets/import"
        WITH DESTINATION_TOPIC "analytics/domo/datasets/import/response"
        WITH METHOD "PUT"
        WITH ENDPOINT "/v1/datasets/{value.json.dataset_id}/data?updateMethod={value.json.update_method}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "User-Agent" "Coreflux-Broker REST_API route (https://www.coreflux.org)"
            ADD "Authorization" "Bearer {value.json.access_token}"
            ADD "Content-Type" "text/csv"
        WITH BODY "{value.json.csv}"
    ADD EVENT DatasetQuery
        WITH SOURCE_TOPIC "analytics/domo/datasets/query"
        WITH DESTINATION_TOPIC "analytics/domo/datasets/query/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/v1/datasets/query/execute/{value.json.dataset_id}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "User-Agent" "Coreflux-Broker REST_API route (https://www.coreflux.org)"
            ADD "Authorization" "Bearer {value.json.access_token}"
            ADD "Content-Type" "application/json"
        WITH BODY "{value.json.body}"
    ADD EVENT SimpleImport
        WITH SOURCE_TOPIC "analytics/domo/simple/import"
        WITH DESTINATION_TOPIC "analytics/domo/simple/import/response"
        WITH METHOD "PUT"
        WITH ENDPOINT "/v1/json/{value.json.dataset_id}/data?updateMethod={value.json.update_method}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "User-Agent" "Coreflux-Broker REST_API route (https://www.coreflux.org)"
            ADD "Authorization" "Bearer {value.json.access_token}"
            ADD "Content-Type" "application/json"
        WITH BODY "{value.json.body}"
    ADD EVENT StreamExecutionCreate
        WITH SOURCE_TOPIC "analytics/domo/streams/executions/create"
        WITH DESTINATION_TOPIC "analytics/domo/streams/executions/create/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/v1/streams/{value.json.stream_id}/executions"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "User-Agent" "Coreflux-Broker REST_API route (https://www.coreflux.org)"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY ""
    ADD EVENT StreamPartUpload
        WITH SOURCE_TOPIC "analytics/domo/streams/executions/part"
        WITH DESTINATION_TOPIC "analytics/domo/streams/executions/part/response"
        WITH METHOD "PUT"
        WITH ENDPOINT "/v1/streams/{value.json.stream_id}/executions/{value.json.execution_id}/part/{value.json.part_id}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "User-Agent" "Coreflux-Broker REST_API route (https://www.coreflux.org)"
            ADD "Authorization" "Bearer {value.json.access_token}"
            ADD "Content-Type" "text/csv"
        WITH BODY "{value.json.csv}"
    ADD EVENT StreamExecutionCommit
        WITH SOURCE_TOPIC "analytics/domo/streams/executions/commit"
        WITH DESTINATION_TOPIC "analytics/domo/streams/executions/commit/response"
        WITH METHOD "PUT"
        WITH ENDPOINT "/v1/streams/{value.json.stream_id}/executions/{value.json.execution_id}/commit"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "User-Agent" "Coreflux-Broker REST_API route (https://www.coreflux.org)"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY ""

Examples

Keep a token for the other actions

DEFINE ACTION DomoToken
ON EVERY 50 MINUTES DO
    SET "auth" WITH GET SECRET "DOMO_BASIC_AUTH"
    PUBLISH TOPIC "analytics/domo/auth/token" WITH { "basic_auth": {auth}, "scope": "data%20user" }

DEFINE ACTION DomoKeepToken
ON TOPIC "analytics/domo/auth/token/response" DO
    SET "token" WITH GET JSON "access_token" IN PAYLOAD AS STRING
    KEEP TOPIC "state/domo/access_token" WITH {token}

Append OEE samples to a DataSet

The Simple API takes JSON rows and converts them to CSV. Create the DataSet once with SimpleDatasetCreate (the first import sets the schema) or with DatasetCreate and an explicit schema, then append:

DEFINE ACTION DomoAppendOee
ON TOPIC "plant/line2/oee" DO
    SET "token" WITH GET TOPIC "state/domo/access_token" AS STRING
    SET "ts" WITH GET JSON "ts" IN PAYLOAD AS STRING
    SET "oee" WITH GET JSON "oee" IN PAYLOAD AS DOUBLE
    PUBLISH TOPIC "analytics/domo/simple/import" WITH { "access_token": {token}, "dataset_id": "0c1e0dbe-9f71-4625-9b50-b79e6e4266f2", "update_method": "APPEND", "body": [ { "ts": {ts}, "line": "Line 2", "oee": {oee} } ] }

The same rows as CSV, published to analytics/domo/datasets/import (Domo answers 204 with an empty body):

{"access_token": "<access_token>", "dataset_id": "0c1e0dbe-9f71-4625-9b50-b79e6e4266f2", "update_method": "APPEND", "csv": "2026-10-05T10:00:00,Line 2,0.87\n2026-10-05T10:01:00,Line 2,0.86\n"}

Bulk load with the Stream API

For large or frequent loads, create a Stream once with StreamCreate (it creates its own DataSet; Stream imports only work on DataSets created this way). Then for each load:

Publish {"access_token": "<access_token>", "stream_id": "42"} to analytics/domo/streams/executions/create. The reply carries the execution id.
Publish each part to analytics/domo/streams/executions/part with part_id 1, 2, 3 and so on. Parts can be sent in parallel; retry a failed part with the same part_id.

{"access_token": "<access_token>", "stream_id": "42", "execution_id": "1", "part_id": "1", "csv": "2026-10-05T10:00:00,L2.Temp,71.5\n2026-10-05T10:00:00,L2.Pressure,3.2\n"}

Publish {"access_token": "<access_token>", "stream_id": "42", "execution_id": "1"} to analytics/domo/streams/executions/commit. Domo imports all parts at once. Use analytics/domo/streams/executions/abort instead to drop the parts.

Query a DataSet and publish the rows

Publish to analytics/domo/datasets/query. The table is always called table. The reply has columns, metadata and rows:

{"access_token": "<access_token>", "dataset_id": "0c1e0dbe-9f71-4625-9b50-b79e6e4266f2", "body": {"sql": "SELECT line, AVG(oee) AS oee FROM table GROUP BY line"}}

Notes

Use DatasetImport or SimpleImport for small DataSets that are updated now and then, and the Stream API for large or fast-growing DataSets, as Domo recommends. REPLACE replaces every row; the schema must match on every import.
StreamsSearch with q dataSource.id:<dataset_id> finds the Stream of a DataSet. StreamExecutionGet shows currentState (ACTIVE, SUCCESS, ERROR, ABORTED).
DatasetExport returns CSV (Accept: text/csv), published as received. For large DataSets query only the rows you need with DatasetQuery.
CardEmbedToken needs a card embed id from Domo Everywhere and returns a token for the embed page. Treat the reply like a credential.
List events default to 50 items (limit up to 50 for DataSets, 500 for users and Streams); page with offset.
Not packaged on purpose: every DELETE (DataSets, Streams, PDP policies, groups, pages, collections, group members), user, group and page writes, PDP policy writes, the Account and Projects APIs, gzip-compressed parts, and the instance Product APIs (<instance>.domo.com/api/... with developer tokens).
Domo’s data platform business is now part of Progress Software; the Platform API host and the OAuth flow are unchanged.

Status

Technical preview, blocked on a live test. Static, spec, example-body and mock checks pass, and keyless calls to api.domo.com return the documented 401 answers. A live run needs a Domo developer client (client ID and secret with the data and user scopes) and a test DataSet.

Quick Start

Coreflux MQTT Broker

LoT Language

Coreflux HUB

VS Code - LoT Notebooks

Coreflux Portal

Using AI with Coreflux

Product Updates

Overview

Setup

Authentication

Events

Payloads

Basic Syntax

Examples

Keep a token for the other actions

Append OEE samples to a DataSet

Bulk load with the Stream API

Query a DataSet and publish the rows

Notes

Status

​Overview

​Setup

​Authentication

​Events

​Payloads

​Basic Syntax

​Examples

​Keep a token for the other actions

​Append OEE samples to a DataSet

​Bulk load with the Stream API

​Query a DataSet and publish the rows

​Notes

​Status

Overview

Setup

Authentication

Events

Payloads

Basic Syntax

Examples

Keep a token for the other actions

Append OEE samples to a DataSet

Bulk load with the Stream API

Query a DataSet and publish the rows

Notes

Status