> ## Documentation Index
> Fetch the complete documentation index at: https://docs.coreflux.org/llms.txt
> Use this file to discover all available pages before exploring further.

# Domo

> Client-mode REST API connector for Domo: push plant data into Domo DataSets as CSV or JSON rows or through the Stream API in parts, query a DataSet with SQL and publish the rows on MQTT, and look up streams, users, groups, pages and the activity log, authenticated with OAuth 2.0 client credentials

Technical Preview

## Overview

A client-mode connector for the [Domo Platform APIs](https://developer.domo.com/portal/8ba9aedad3679-ap-is) (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](/latest/lot-language/routes/rest-api-routes).

## Setup

1. Sign in at [developer.domo.com](https://developer.domo.com) with a Domo account of your instance and open **My Account > New Client**.
2. 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**.
3. 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).
4. 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.

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{"basic_auth": "<DOMO_BASIC_AUTH>", "scope": "data%20user"}
```

> [!WARNING]
>
> `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):

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
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

```lot theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
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:

```lot theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
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):

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{"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:

1. Publish `{"access_token": "<access_token>", "stream_id": "42"}` to `analytics/domo/streams/executions/create`. The reply carries the execution `id`.
2. 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`.

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{"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"}
```

3. 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`:

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{"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.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.