Skip to content

dataset

dataset

Dataset route module - re-exports all dataset functionality.

UploadDataError 

UploadDataError(
    stage_num: int,
    dataset_id: str,
    res: ResponseGetData,
    message: str | None = None,
)

Bases: RouteError

raise if unable to upload data to Domo

Source code in src/crew_dcs/routes/dataset/exceptions.py 
54
55
56
57
58
59
60
61
62
63
def __init__(
    self,
    stage_num: int,
    dataset_id: str,
    res: rgd.ResponseGetData,
    message: str | None = None,
):
    message = f"error uploading data during Stage {stage_num} - {message}"

    super().__init__(entity_id=dataset_id, message=message, res=res)

alter_schema async 

alter_schema(
    auth: DomoAuth,
    schema_obj: dict,
    dataset_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

alters the schema for a dataset BUT DOES NOT ALTER THE DESCRIPTION

Source code in src/crew_dcs/routes/dataset/schema.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def alter_schema(
    auth: DomoAuth,
    schema_obj: dict,
    dataset_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """alters the schema for a dataset BUT DOES NOT ALTER THE DESCRIPTION"""

    url = f"https://{auth.domo_instance}.domo.com/api/data/v2/datasources/{dataset_id}/schemas"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=schema_obj,
        context=context,
    )

    if not res.is_success:
        raise Dataset_CRUD_Error(dataset_id=dataset_id, res=res)

    return res

alter_schema_descriptions async 

alter_schema_descriptions(
    auth: DomoAuth,
    schema_obj: dict,
    dataset_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

alters the description of the schema columns // as seen in DataCenter > Dataset > Schema

Source code in src/crew_dcs/routes/dataset/schema.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def alter_schema_descriptions(
    auth: DomoAuth,
    schema_obj: dict,
    dataset_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """alters the description of the schema columns // as seen in DataCenter > Dataset > Schema"""

    url = f"https://{auth.domo_instance}.domo.com/api/query/v1/datasources/{dataset_id}/wrangle"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=schema_obj,
        context=context,
    )

    if not res.is_success:
        raise Dataset_CRUD_Error(dataset_id=dataset_id, res=res)

    return res

create_semantic_model async 

create_semantic_model(
    auth: DomoAuth,
    data_source_name: str,
    schema: dict,
    *,
    data_source_description: str | None = None,
    cloud_id: str = "domo",
    body: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create a new data model (POST /api/query/v1/semantic-models).

Parameters:

Name Type Description Default
auth DomoAuth

Authentication object

 required
data_source_name str

Model display name

 required
schema dict

The model definition: {"objects": {...}, "relationships": [...]} (the shape produced by DataModelTemplate.to_schema())

 required
data_source_description str | None

Optional model description

 None
cloud_id str

Cloud engine ID (defaults to "domo")

 'domo'
body dict | None

Fully pre-built payload; overrides the generated envelope

 None
context RouteContext | None

RouteContext for request configuration

 None
**context_kwargs

Additional context parameters

 {}

Returns:

Type Description
ResponseGetData

ResponseGetData containing the new model's dataSourceId.
Note

The Domo API has a validation bug where non-primary relationships require a non-null alias field but rejects all string values for it. The workaround is to set primary=True for all relationships in the schema. The DataModelTemplate.to_schema() method preserves whatever primary values are set on the relationships.

Source code in src/crew_dcs/routes/dataset/dataset_views.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_semantic_model(
    auth: DomoAuth,
    data_source_name: str,
    schema: dict,
    *,
    data_source_description: str | None = None,
    cloud_id: str = "domo",
    body: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a new data model (POST /api/query/v1/semantic-models).

    Args:
        auth: Authentication object
        data_source_name: Model display name
        schema: The model definition: ``{"objects": {...}, "relationships": [...]}``
            (the shape produced by ``DataModelTemplate.to_schema()``)
        data_source_description: Optional model description
        cloud_id: Cloud engine ID (defaults to "domo")
        body: Fully pre-built payload; overrides the generated envelope
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters

    Returns:
        ResponseGetData containing the new model's ``dataSourceId``.

    Note:
        The Domo API has a validation bug where non-primary relationships
        require a non-null ``alias`` field but rejects all string values for it.
        The workaround is to set ``primary=True`` for all relationships in the
        schema. The ``DataModelTemplate.to_schema()`` method preserves whatever
        ``primary`` values are set on the relationships.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/query/v1/semantic-models"

    if body is None:
        body = {
            "dataSourceName": data_source_name,
            "dataSourceDescription": data_source_description,
            "cloudId": cloud_id,
            "schema": schema,
        }

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Dataset_CRUD_Error(dataset_id="new", res=res)

    return res

delete_partition_stage_1 async 

delete_partition_stage_1(
    auth: DomoAuth,
    dataset_id: str,
    dataset_partition_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Delete partition has 3 stages

Stage 1. This marks the data version associated with the partition tag as deleted.

It does not delete the partition tag or remove the association between the partition tag and data version. There should be no need to upload an empty file - step #3 will remove the data from Adrenaline.

update on 9/9/2022 based on the conversation with Greg Swensen

Source code in src/crew_dcs/routes/dataset/core.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def delete_partition_stage_1(
    auth: DomoAuth,
    dataset_id: str,
    dataset_partition_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Delete partition has 3 stages
    # Stage 1. This marks the data version associated with the partition tag as deleted.
    It does not delete the partition tag or remove the association between the partition tag and data version.
    There should be no need to upload an empty file - step #3 will remove the data from Adrenaline.
    # update on 9/9/2022 based on the conversation with Greg Swensen"""

    url = f"https://{auth.domo_instance}.domo.com/api/query/v1/datasources/{dataset_id}/tag/{dataset_partition_id}/data"

    res = await gd.get_data(
        auth=auth,
        method="DELETE",
        url=url,
        context=context,
    )

    if not res.is_success:
        raise Dataset_CRUD_Error(dataset_id=dataset_id, res=res)

    return res

delete_partition_stage_2 async 

delete_partition_stage_2(
    auth: DomoAuth,
    dataset_id: str,
    dataset_partition_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

This will remove the partition association so that it doesn't show up in the list call. Technically, this is not required as a partition against a deleted data version will not count against the 400 partition limit but as the current partitions api doesn't make that clear, cleaning these up will make it much easier for you to manage.

Source code in src/crew_dcs/routes/dataset/core.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def delete_partition_stage_2(
    auth: DomoAuth,
    dataset_id: str,
    dataset_partition_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """This will remove the partition association so that it doesn't show up in the list call.
    Technically, this is not required as a partition against a deleted data version will not count against the 400 partition limit
    but as the current partitions api doesn't make that clear, cleaning these up will make it much easier for you to manage.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/query/v1/datasources/{dataset_id}/partition/{dataset_partition_id}"

    res = await gd.get_data(
        auth=auth,
        method="DELETE",
        url=url,
        context=context,
    )

    if not res.is_success:
        raise Dataset_CRUD_Error(dataset_id=dataset_id, res=res)

    return res

generate_schema_from_dataframe 

generate_schema_from_dataframe(
    df: DataFrame,
    upsert_key: str | None = None,
    column_types: dict[str, str] | None = None,
) -> dict

Generate a Domo schema dict from a pandas DataFrame.

Maps pandas dtypes to Domo schema column types: object/bool -> STRING, integer -> LONG, float -> DOUBLE, datetime -> DATETIME.

Parameters:

Name Type Description Default
df DataFrame

pandas DataFrame to analyze

 required
upsert_key str | None

Column name to mark as the identity/upsert column

 None
column_types dict[str, str] | None

Override mapping {column_name: "STRING"|"LONG"|"DOUBLE"|"DATE"|"DATETIME"}

 None

Returns:

Type Description
dict

{"columns": [{"name": "...", "type": "...", "upsertKey": bool}, ...]}
Source code in src/crew_dcs/routes/dataset/convenience.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def generate_schema_from_dataframe(
    df: pd.DataFrame,
    upsert_key: str | None = None,
    column_types: dict[str, str] | None = None,
) -> dict:
    """Generate a Domo schema dict from a pandas DataFrame.

    Maps pandas dtypes to Domo schema column types:
    object/bool -> STRING, integer -> LONG, float -> DOUBLE, datetime -> DATETIME.

    Args:
        df: pandas DataFrame to analyze
        upsert_key: Column name to mark as the identity/upsert column
        column_types: Override mapping {column_name: "STRING"|"LONG"|"DOUBLE"|"DATE"|"DATETIME"}

    Returns:
        {"columns": [{"name": "...", "type": "...", "upsertKey": bool}, ...]}
    """
    column_types = column_types or {}

    columns = []
    for col_name in df.columns:
        col_type = column_types.get(col_name) or _detect_domo_type(df[col_name])
        columns.append(
            {
                "name": col_name,
                "type": _DOMO_TYPE_MAP.get(col_type, col_type),
                "upsertKey": col_name == upsert_key,
            }
        )

    return {"columns": columns}

generate_share_dataset_payload 

generate_share_dataset_payload(
    entity_type: str,
    entity_id: str,
    access_level: ShareDataset_AccessLevelEnum = CAN_SHARE,
    is_send_email: bool = False,
) -> dict

Generate payload for sharing a dataset.

Parameters:

Name Type Description Default
entity_type str

Type of entity (USER or GROUP)

 required
entity_id str

ID of the user or group

 required
access_level ShareDataset_AccessLevelEnum

Access level (ShareDataset_AccessLevelEnum enum). Defaults to CAN_SHARE.

 CAN_SHARE
is_send_email bool

Whether to send email notification

 False

Returns:

Type Description
dict

Dictionary payload for dataset sharing API
Source code in src/crew_dcs/routes/dataset/sharing.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def generate_share_dataset_payload(
    entity_type: str,  # USER or GROUP
    entity_id: str,
    access_level: ShareDataset_AccessLevelEnum = ShareDataset_AccessLevelEnum.CAN_SHARE,
    is_send_email: bool = False,
) -> dict:
    """Generate payload for sharing a dataset.

    Args:
        entity_type: Type of entity (USER or GROUP)
        entity_id: ID of the user or group
        access_level: Access level (ShareDataset_AccessLevelEnum enum). Defaults to CAN_SHARE.
        is_send_email: Whether to send email notification

    Returns:
        Dictionary payload for dataset sharing API
    """
    return {
        "permissions": [
            {"type": entity_type, "id": entity_id, "accessLevel": access_level.value}
        ],
        "sendEmail": is_send_email,
    }

get_dataset_by_id async 

get_dataset_by_id(
    dataset_id: str,
    auth: DomoAuth | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

retrieve dataset metadata

Source code in src/crew_dcs/routes/dataset/core.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_dataset_by_id(
    dataset_id: str,  # dataset id from URL
    auth: DomoAuth | None = None,  # requires full authentication
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:  # returns metadata about a dataset
    """retrieve dataset metadata"""

    url = f"https://{auth.domo_instance}.domo.com/api/data/v3/datasources/{dataset_id}"  # type: ignore

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        context=context,
    )

    if res.status == 404 and res.response == "Not Found":
        raise DatasetNotFoundError(dataset_id=dataset_id, res=res)

    if not res.is_success:
        raise Dataset_GET_Error(dataset_id=dataset_id, res=res)

    return res

get_dataset_view_schema_indexed async 

get_dataset_view_schema_indexed(
    auth: DomoAuth,
    dataset_id: str,
    include_data_control_column_details: bool = True,
    flatten: bool | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Get the indexed schema for a dataset view with optional data control column details.

This endpoint returns the schema structure for a dataset view, including: - Column definitions with types and visibility - SELECT query structure - View template information - Data control column details (if requested)

Parameters:

Name Type Description Default
auth DomoAuth

Authentication object

 required
dataset_id str

The dataset view ID

 required
include_data_control_column_details bool

If True, includes data control column details

 True
flatten bool | None

If False, preserves per-table join structure (required to get a data model's model.objects / modeledColumns). If None, the param is omitted and the API default applies.

 None
context RouteContext | None

RouteContext for request configuration

 None
**context_kwargs

Additional context parameters (session, debug_api, etc.)

 {}

Returns:

Type Description
ResponseGetData

ResponseGetData object containing:
ResponseGetData
  • name: View name
ResponseGetData
  • tables: Array of tables with column definitions
ResponseGetData
  • select: SELECT query structure
ResponseGetData
  • dataSourceId: Dataset ID
ResponseGetData
  • versionId: Version ID
ResponseGetData
  • viewTemplate: Template information with select string and fromItemInfo
ResponseGetData
  • dataControlDetails: Data control column details (if requested)

Raises:

Type Description
Dataset_GET_Error

If schema retrieval fails
Source code in src/crew_dcs/routes/dataset/dataset_views.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_dataset_view_schema_indexed(
    auth: DomoAuth,
    dataset_id: str,
    include_data_control_column_details: bool = True,
    flatten: bool | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get the indexed schema for a dataset view with optional data control column details.

    This endpoint returns the schema structure for a dataset view, including:
    - Column definitions with types and visibility
    - SELECT query structure
    - View template information
    - Data control column details (if requested)

    Args:
        auth: Authentication object
        dataset_id: The dataset view ID
        include_data_control_column_details: If True, includes data control column details
        flatten: If False, preserves per-table join structure (required to get a
            data model's ``model.objects`` / ``modeledColumns``). If None, the
            param is omitted and the API default applies.
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing:
        - name: View name
        - tables: Array of tables with column definitions
        - select: SELECT query structure
        - dataSourceId: Dataset ID
        - versionId: Version ID
        - viewTemplate: Template information with select string and fromItemInfo
        - dataControlDetails: Data control column details (if requested)

    Raises:
        Dataset_GET_Error: If schema retrieval fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/query/v1/datasources/{dataset_id}/schema/indexed"

    params = {}
    if include_data_control_column_details:
        params["options"] = "INCLUDE_DATA_CONTROL_COLUMN_DETAILS"
    if flatten is not None:
        params["flatten"] = "true" if flatten else "false"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        params=params,
        context=context,
    )

    if not res.is_success:
        raise Dataset_GET_Error(dataset_id=dataset_id, res=res)

    return res

get_datasets_by_ids async 

get_datasets_by_ids(
    auth: DomoAuth,
    dataset_ids: list[str],
    include_private: bool = True,
    include_all_details: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Fetch metadata for many datasources at once.

POST /api/data/v3/datasources/bulk (body = list of dataset ids). The API returns {"dataSources": [...], "_metaData": {...}}; unless return_raw is set, res.response is unwrapped to the dataSources list.

Source code in src/crew_dcs/routes/dataset/core.py 
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_datasets_by_ids(
    auth: DomoAuth,
    dataset_ids: list[str],
    include_private: bool = True,
    include_all_details: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Fetch metadata for many datasources at once.

    POST /api/data/v3/datasources/bulk (body = list of dataset ids). The API
    returns ``{"dataSources": [...], "_metaData": {...}}``; unless ``return_raw``
    is set, ``res.response`` is unwrapped to the ``dataSources`` list.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    params = {
        "includePrivate": str(include_private).lower(),
        "includeAllDetails": str(include_all_details).lower(),
    }
    url = f"https://{auth.domo_instance}.domo.com/api/data/v3/datasources/bulk"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=list(dataset_ids),
        params=params,
        context=context,
    )

    if not res.is_success:
        raise Dataset_GET_Error(res=res)

    if not return_raw and isinstance(res.response, dict):
        res.response = res.response.get("dataSources", res.response)

    return res

get_permissions async 

get_permissions(
    auth: DomoAuth,
    dataset_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

retrieve the schema for a dataset

Source code in src/crew_dcs/routes/dataset/sharing.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_permissions(
    auth: DomoAuth,
    dataset_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """retrieve the schema for a dataset"""

    url = f"https://{auth.domo_instance}.domo.com/api/data/v3/datasources/{dataset_id}/permissions"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        context=context,
    )

    if not res.is_success:
        raise Dataset_GET_Error(dataset_id=dataset_id, res=res)

    return res

get_schema async 

get_schema(
    auth: DomoAuth,
    dataset_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

retrieve the schema for a dataset

Source code in src/crew_dcs/routes/dataset/schema.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_schema(
    auth: DomoAuth,
    dataset_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """retrieve the schema for a dataset"""

    url = f"https://{auth.domo_instance}.domo.com/api/query/v1/datasources/{dataset_id}/schema/indexed?includeHidden=false"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        context=context,
    )

    if not res.is_success:
        raise Dataset_GET_Error(dataset_id=dataset_id, res=res)

    return res

index_dataset async 

index_dataset(
    auth: DomoAuth,
    dataset_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

manually index a dataset

Source code in src/crew_dcs/routes/dataset/upload.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def index_dataset(
    auth: DomoAuth,
    dataset_id: str,
    *,  # Make following params keyword-only
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """manually index a dataset"""

    url = f"https://{auth.domo_instance}.domo.com/api/data/v3/datasources/{dataset_id}/indexes"

    body = {"dataIds": []}

    res = await gd.get_data(
        auth=auth,
        method="POST",
        body=body,
        url=url,
        context=context,
    )

    if not res.is_success:
        raise Dataset_CRUD_Error(dataset_id=dataset_id, res=res)

    return res

index_status async 

index_status(
    auth: DomoAuth,
    dataset_id: str,
    index_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

get the completion status of an index

Source code in src/crew_dcs/routes/dataset/upload.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def index_status(
    auth: DomoAuth,
    dataset_id: str,
    index_id: str,
    *,  # Make following params keyword-only
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """get the completion status of an index"""

    url = f"https://{auth.domo_instance}.domo.com/api/data/v3/datasources/{dataset_id}/indexes/{index_id}/statuses"

    res = await gd.get_data(
        auth=auth,
        method="GET",
        url=url,
        context=context,
    )

    if not res.is_success:
        raise Dataset_GET_Error(dataset_id=dataset_id, res=res)

    return res

list_partitions async 

list_partitions(
    auth: DomoAuth,
    dataset_id: str,
    body: dict | None = None,
    debug_loop: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

List all partitions for a dataset.

Source code in src/crew_dcs/routes/dataset/upload.py 
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def list_partitions(
    auth: DomoAuth,
    dataset_id: str,
    body: dict | None = None,
    debug_loop: bool = False,
    *,  # Make following params keyword-only
    context: RouteContext | None = None,
    **context_kwargs,
):
    """List all partitions for a dataset."""

    body = body or generate_list_partitions_body()

    url = f"https://{auth.domo_instance}.domo.com/api/query/v1/datasources/{dataset_id}/partition/list"

    offset_params = {
        "offset": "offset",
        "limit": "limit",
    }

    def arr_fn(res) -> list[dict]:
        return res.response

    res = await gd.looper(
        auth=auth,
        method="POST",
        url=url,
        arr_fn=arr_fn,
        body=body,
        offset_params_in_body=True,
        offset_params=offset_params,
        loop_until_end=True,
        debug_loop=debug_loop,
        context=context,
    )

    if res.status == 404 and res.response == "Not Found":
        raise DatasetNotFoundError(dataset_id=dataset_id, res=res)

    if not res.is_success:
        raise Dataset_GET_Error(dataset_id=dataset_id, res=res)

    return res

query_dataset_public async 

query_dataset_public(
    dev_auth: DomoDeveloperAuth,
    dataset_id: str,
    sql: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

query for hitting public apis, requires client_id and secret authentication

Source code in src/crew_dcs/routes/dataset/query.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def query_dataset_public(
    dev_auth: dmda.DomoDeveloperAuth,
    dataset_id: str,
    sql: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """query for hitting public apis, requires client_id and secret authentication"""

    url = f"https://api.domo.com/v1/datasets/query/execute/{dataset_id}?IncludeHeaders=true"

    body = {"sql": sql}

    res = await gd.get_data(
        auth=dev_auth,
        url=url,
        method="POST",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Dataset_CRUD_Error(dataset_id=dataset_id, res=res)

    return res

search_datasets async 

search_datasets(
    auth: DomoAuth,
    search_text: str | None = None,
    maximum: int | None = None,
    data_provider_type: str | None = None,
    tags: list[str] | str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Search for datasets by name, data provider, and/or tags.

Uses the datacenter search API to find datasets matching the search criteria.

Parameters:

Name Type Description Default
auth DomoAuth

Authentication object

 required
search_text str | None

Optional text to search for in dataset names (wildcards supported)

 None
maximum int | None

Maximum number of results to return

 None
data_provider_type str | None

Optional dataset provider type filter

 None
tags list[str] | str | None

Optional tag filter(s) — a single tag string or a list of tag strings. Each tag generates a tag_facet term filter.

 None
return_raw bool

Return raw response without processing

 False
context RouteContext | None

RouteContext for request configuration

 None
**context_kwargs

Additional context parameters (session, debug_api, etc.)

 {}

Returns:

Type Description
ResponseGetData

ResponseGetData object containing list of matching datasets

Raises:

Type Description
Dataset_GET_Error

If search operation fails
Source code in src/crew_dcs/routes/dataset/core.py 
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def search_datasets(
    auth: DomoAuth,
    search_text: str | None = None,
    maximum: int | None = None,
    data_provider_type: str | None = None,
    tags: list[str] | str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Search for datasets by name, data provider, and/or tags.

    Uses the datacenter search API to find datasets matching the search criteria.

    Args:
        auth: Authentication object
        search_text: Optional text to search for in dataset names (wildcards supported)
        maximum: Maximum number of results to return
        data_provider_type: Optional dataset provider type filter
        tags: Optional tag filter(s) — a single tag string or a list of tag strings.
            Each tag generates a ``tag_facet`` term filter.
        return_raw: Return raw response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing list of matching datasets

    Raises:
        Dataset_GET_Error: If search operation fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    from ..datacenter import (
        Datacenter_Enum,
        Datacenter_Filter_Field_Enum,
        generate_data_center_profile_body,
        generate_search_datacenter_filter,
        generate_search_datacenter_filters,
        generate_search_datacenter_filter_search_term,
        search_datacenter,
    )
    from ..datacenter.exceptions import SearchDatacenterNoResultsFoundError

    # Normalize tags to a list
    if tags is not None and isinstance(tags, str):
        tags = [tags]

    try:
        body = None
        if data_provider_type or tags:
            # Build declarative (value, builder) filter specs.
            # generate_search_datacenter_filters skips entries whose value is None.
            filter_specs = [
                (
                    search_text,
                    lambda value: generate_search_datacenter_filter_search_term(value),
                ),
                (
                    data_provider_type,
                    lambda value: {
                        **generate_search_datacenter_filter(
                            Datacenter_Filter_Field_Enum.DATAPROVIDER,
                            value,
                        ),
                        "name": value,
                    },
                ),
            ]

            # Add one tag filter per tag
            if tags:
                for tag in tags:
                    filter_specs.append(
                        (
                            tag,
                            lambda value: generate_search_datacenter_filter(
                                Datacenter_Filter_Field_Enum.TAG,
                                value,
                            ),
                        )
                    )

            filters = generate_search_datacenter_filters(filter_specs)

            body = generate_data_center_profile_body(
                maximum=maximum,
                filters=filters,
                entity_type=Datacenter_Enum.DATASET,
            )

        res = await search_datacenter(
            auth=auth,
            body=body,
            search_text=search_text,
            entity_type=Datacenter_Enum.DATASET,
            maximum=maximum,
            context=context,
        )
    except SearchDatacenterNoResultsFoundError:
        # No results is valid - return empty list
        res = rgd.ResponseGetData(
            status=200,
            response=[],
            is_success=True,
        )

    if return_raw:
        return res

    if not res.is_success:
        raise Dataset_GET_Error(res=res)

    return res

set_dataset_tags async 

set_dataset_tags(
    auth: DomoAuth,
    tag_ls: list[str],
    dataset_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

REPLACE tags on this dataset with a new list

Source code in src/crew_dcs/routes/dataset/schema.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def set_dataset_tags(
    auth: DomoAuth,
    tag_ls: list[str],  # complete list of tags for dataset
    dataset_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """REPLACE tags on this dataset with a new list"""

    url = f"https://{auth.domo_instance}.domo.com/api/data/ui/v3/datasources/{dataset_id}/tags"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=tag_ls,
        return_raw=return_raw,
        context=context,
    )

    if return_raw:
        return res

    if res.status == 200:
        res.response = f"Dataset {dataset_id} tags updated to [{', '.join(tag_ls)}]"

    if not res.is_success:
        raise Dataset_CRUD_Error(dataset_id=dataset_id, res=res)

    return res

update_dataset_properties async 

update_dataset_properties(
    auth: DomoAuth,
    dataset_id: str,
    name: str | None = None,
    description: str | None = None,
    body: dict | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update a datasource's name and/or description.

PUT /api/data/v3/datasources/{dataset_id}/properties

Source code in src/crew_dcs/routes/dataset/core.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def update_dataset_properties(
    auth: DomoAuth,
    dataset_id: str,
    name: str | None = None,
    description: str | None = None,
    body: dict | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update a datasource's name and/or description.

    PUT /api/data/v3/datasources/{dataset_id}/properties
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/data/v3/datasources/{dataset_id}/properties"

    if body is None:
        body = {"dataSourceName": name, "dataSourceDescription": description}

    res = await gd.get_data(
        auth=auth, url=url, method="PUT", body=body, context=context
    )

    if not res.is_success:
        raise Dataset_CRUD_Error(dataset_id=dataset_id, res=res)

    return res

update_semantic_model async 

update_semantic_model(
    auth: DomoAuth,
    model_id: str,
    schema: dict,
    *,
    data_source_name: str | None = None,
    data_source_description: str | None = None,
    last_updated: str | None = None,
    cloud_id: str = "domo",
    can_edit: bool = True,
    body: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update a data model (PUT /api/query/v1/semantic-models/{model_id}).

Parameters:

Name Type Description Default
auth DomoAuth

Authentication object

 required
model_id str

The data model (semantic model) datasource ID

 required
schema dict

The model definition: {"objects": {...}, "relationships": [...]} (the shape produced by DataModelTemplate.to_schema())

 required
data_source_name str | None

Model display name

 None
data_source_description str | None

Model description

 None
last_updated str | None

ISO-8601 last-updated timestamp echoed back to the API

 None
cloud_id str

Cloud engine ID (defaults to "domo")

 'domo'
can_edit bool

Whether the caller can edit (defaults to True)

 True
body dict | None

Fully pre-built payload; overrides the generated envelope

 None
context RouteContext | None

RouteContext for request configuration

 None
**context_kwargs

Additional context parameters

 {}
Source code in src/crew_dcs/routes/dataset/dataset_views.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def update_semantic_model(
    auth: DomoAuth,
    model_id: str,
    schema: dict,
    *,
    data_source_name: str | None = None,
    data_source_description: str | None = None,
    last_updated: str | None = None,
    cloud_id: str = "domo",
    can_edit: bool = True,
    body: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update a data model (PUT /api/query/v1/semantic-models/{model_id}).

    Args:
        auth: Authentication object
        model_id: The data model (semantic model) datasource ID
        schema: The model definition: ``{"objects": {...}, "relationships": [...]}``
            (the shape produced by ``DataModelTemplate.to_schema()``)
        data_source_name: Model display name
        data_source_description: Model description
        last_updated: ISO-8601 last-updated timestamp echoed back to the API
        cloud_id: Cloud engine ID (defaults to "domo")
        can_edit: Whether the caller can edit (defaults to True)
        body: Fully pre-built payload; overrides the generated envelope
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = (
        f"https://{auth.domo_instance}.domo.com/api/query/v1/semantic-models/{model_id}"
    )

    if body is None:
        body = {
            "dataSourceName": data_source_name,
            "dataSourceDescription": data_source_description,
            "lastUpdated": last_updated,
            "schema": schema,
            "cloudId": cloud_id,
            "canEdit": can_edit,
        }

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="PUT",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Dataset_CRUD_Error(dataset_id=model_id, res=res)

    return res

update_view async 

update_view(
    auth: DomoAuth,
    view_id: str,
    schema: dict,
    *,
    data_source_name: str | None = None,
    trigger: dict | None = None,
    data_provider_type: str | None = None,
    body: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update a dataset view's definition (PUT /api/query/v1/views/{view_id}).

Parameters:

Name Type Description Default
auth DomoAuth

Authentication object

 required
view_id str

The dataset view ID

 required
schema dict

The view schema payload (tables + viewTemplate + tableAliases). Typically a (possibly edited) get_dataset_view_schema_indexed response augmented with tableAliases.

 required
data_source_name str | None

View display name

 None
trigger dict | None

Refresh trigger config (defaults to {})

 None
data_provider_type str | None

Data provider type (usually None for views)

 None
body dict | None

Fully pre-built payload; overrides the generated envelope

 None
context RouteContext | None

RouteContext for request configuration

 None
**context_kwargs

Additional context parameters

 {}
Source code in src/crew_dcs/routes/dataset/dataset_views.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def update_view(
    auth: DomoAuth,
    view_id: str,
    schema: dict,
    *,
    data_source_name: str | None = None,
    trigger: dict | None = None,
    data_provider_type: str | None = None,
    body: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update a dataset view's definition (PUT /api/query/v1/views/{view_id}).

    Args:
        auth: Authentication object
        view_id: The dataset view ID
        schema: The view schema payload (tables + viewTemplate + tableAliases).
            Typically a (possibly edited) ``get_dataset_view_schema_indexed``
            response augmented with ``tableAliases``.
        data_source_name: View display name
        trigger: Refresh trigger config (defaults to ``{}``)
        data_provider_type: Data provider type (usually None for views)
        body: Fully pre-built payload; overrides the generated envelope
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/query/v1/views/{view_id}"

    if body is None:
        body = {
            "dataSourceName": data_source_name,
            "trigger": trigger or {},
            "dataProviderType": data_provider_type,
            "schema": schema,
        }

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="PUT",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Dataset_CRUD_Error(dataset_id=view_id, res=res)

    return res

upload_dataframe async 

upload_dataframe(
    auth: DomoAuth,
    dataset_name: str,
    df: DataFrame,
    *,
    schema: dict | None = None,
    upsert_key: str | None = None,
    column_types: dict[str, str] | None = None,
    update_method: str = "REPLACE",
    is_index: bool = True,
    is_create_if_not_exists: bool = True,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Upload a DataFrame to Domo in one call.

If the dataset doesn't exist, creates it with an auto-detected schema. If it does exist, amends the schema for any new DataFrame columns before uploading.

Parameters:

Name Type Description Default
auth DomoAuth

Authentication object for API requests

 required
dataset_name str

Name of the dataset to find or create

 required
df DataFrame

DataFrame to upload

 required
schema dict | None

Explicit schema override used when creating a new dataset

 None
upsert_key str | None

Column name to mark as the identity/upsert column

 None
column_types dict[str, str] | None

Override auto-detected types for specific columns

 None
update_method str

"REPLACE" or "APPEND"

 'REPLACE'
is_index bool

Whether to index the dataset after uploading

 True
is_create_if_not_exists bool

Create the dataset if no match is found by name

 True
context RouteContext | None

RouteContext for request configuration

 None

Returns:

Type Description
ResponseGetData

ResponseGetData from the upload commit, with dataset_id set.

Raises:

Type Description
DatasetNotFoundError

If no dataset matches and is_create_if_not_exists is False
Source code in src/crew_dcs/routes/dataset/convenience.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def upload_dataframe(
    auth: DomoAuth,
    dataset_name: str,
    df: pd.DataFrame,
    *,
    schema: dict | None = None,
    upsert_key: str | None = None,
    column_types: dict[str, str] | None = None,
    update_method: str = "REPLACE",
    is_index: bool = True,
    is_create_if_not_exists: bool = True,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Upload a DataFrame to Domo in one call.

    If the dataset doesn't exist, creates it with an auto-detected schema.
    If it does exist, amends the schema for any new DataFrame columns before
    uploading.

    Args:
        auth: Authentication object for API requests
        dataset_name: Name of the dataset to find or create
        df: DataFrame to upload
        schema: Explicit schema override used when creating a new dataset
        upsert_key: Column name to mark as the identity/upsert column
        column_types: Override auto-detected types for specific columns
        update_method: "REPLACE" or "APPEND"
        is_index: Whether to index the dataset after uploading
        is_create_if_not_exists: Create the dataset if no match is found by name
        context: RouteContext for request configuration

    Returns:
        ResponseGetData from the upload commit, with `dataset_id` set.

    Raises:
        DatasetNotFoundError: If no dataset matches and is_create_if_not_exists is False
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    dataset_id, search_res = await _find_dataset_id_by_name(
        auth=auth, dataset_name=dataset_name, context=context
    )

    if dataset_id is None:
        if not is_create_if_not_exists:
            raise DatasetNotFoundError(dataset_id=dataset_name, res=search_res)

        create_schema = schema or generate_schema_from_dataframe(
            df, upsert_key=upsert_key, column_types=column_types
        )
        create_res = await create(
            auth=auth,
            dataset_name=dataset_name,
            schema=create_schema,
            context=context,
        )
        dataset_id = create_res.response.get("id") or create_res.response.get(
            "dataSource", {}
        ).get("dataSourceId")
    else:
        schema_res = await get_schema(auth=auth, dataset_id=dataset_id, context=context)
        existing_columns = schema_res.response.get("tables", [{}])[0].get("columns", [])
        existing_names = {col.get("name") for col in existing_columns}
        new_column_names = [name for name in df.columns if name not in existing_names]

        needs_upsert_update = upsert_key is not None and not any(
            col.get("name") == upsert_key and col.get("upsertKey")
            for col in existing_columns
        )

        if new_column_names or needs_upsert_update:
            merged_columns = [
                {
                    "id": col.get("id"),
                    "name": col.get("name"),
                    "type": col.get("type"),
                    "upsertKey": col.get("name") == upsert_key
                    if upsert_key is not None
                    else col.get("upsertKey", False),
                }
                for col in existing_columns
            ]
            for name in new_column_names:
                col_type = (column_types or {}).get(name) or _detect_domo_type(df[name])
                merged_columns.append(
                    {
                        "name": name,
                        "type": _DOMO_TYPE_MAP.get(col_type, col_type),
                        "upsertKey": name == upsert_key,
                    }
                )

            await alter_schema(
                auth=auth,
                dataset_id=dataset_id,
                schema_obj={"columns": merged_columns},
                context=context,
            )

    stage_1_res = await upload_dataset_stage_1(
        auth=auth, dataset_id=dataset_id, context=context
    )
    upload_id = stage_1_res.response

    await upload_dataset_stage_2_df(
        auth=auth,
        dataset_id=dataset_id,
        upload_id=upload_id,
        upload_df=df,
        context=context,
    )

    res = await upload_dataset_stage_3(
        auth=auth,
        dataset_id=dataset_id,
        upload_id=upload_id,
        update_method=update_method,
        is_index=False,
        context=context,
    )

    if is_index:
        res = await index_dataset(auth=auth, dataset_id=dataset_id, context=context)

    res.dataset_id = dataset_id

    return res

upload_dataset_stage_1 async 

upload_dataset_stage_1(
    auth: DomoAuth,
    dataset_id: str,
    partition_tag: str | None = None,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs
) -> ResponseGetData

preps dataset for upload by creating an upload_id (upload session key) pass to stage 2 as a parameter

Source code in src/crew_dcs/routes/dataset/upload.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def upload_dataset_stage_1(
    auth: DomoAuth,
    dataset_id: str,
    partition_tag: str | None = None,  # synonymous with data_tag
    *,  # Make following params keyword-only
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """preps dataset for upload by creating an upload_id (upload session key) pass to stage 2 as a parameter"""

    url = f"https://{auth.domo_instance}.domo.com/api/data/v3/datasources/{dataset_id}/uploads"

    # base body assumes no paritioning
    body = {"action": None, "appendId": None}

    params = None

    if partition_tag:
        params = {"dataTag": partition_tag}
        body.update({"appendId": "latest"})  # type: ignore

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=body,
        params=params,
        context=context,
    )

    if not res.is_success:
        raise UploadDataError(stage_num=1, dataset_id=dataset_id, res=res)

    if return_raw:
        return res

    upload_id = res.response.get("uploadId")

    if not upload_id:
        raise UploadDataError(
            stage_num=1,
            dataset_id=dataset_id,
            res=res,
            message="no upload_id",
        )

    res.response = upload_id

    return res

upload_dataset_stage_2_df async 

upload_dataset_stage_2_df(
    auth: DomoAuth,
    dataset_id: str,
    upload_id: str,
    upload_df: DataFrame,
    part_id: int = 2,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Upload pandas DataFrame to dataset (stage 2 of upload process).

Source code in src/crew_dcs/routes/dataset/upload.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def upload_dataset_stage_2_df(
    auth: DomoAuth,
    dataset_id: str,
    upload_id: str,  # must originate from  a stage_1 upload response
    upload_df: pd.DataFrame,
    part_id: int = 2,  # only necessary if streaming multiple files into the same partition (multi-part upload)
    *,  # Make following params keyword-only
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Upload pandas DataFrame to dataset (stage 2 of upload process)."""

    url = f"https://{auth.domo_instance}.domo.com/api/data/v3/datasources/{dataset_id}/uploads/{upload_id}/parts/{part_id}"

    body = upload_df.to_csv(header=False, index=False)

    # if debug:

    res = await gd.get_data(
        url=url,
        method="PUT",
        auth=auth,
        content_type="text/csv",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise UploadDataError(stage_num=2, dataset_id=dataset_id, res=res)

    res.upload_id = upload_id
    res.dataset_id = dataset_id
    res.part_id = part_id

    return res

upload_dataset_stage_2_file async 

upload_dataset_stage_2_file(
    auth: DomoAuth,
    dataset_id: str,
    upload_id: str,
    data_file: TextIOWrapper | None = None,
    part_id: int = 2,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Upload data file to dataset (stage 2 of upload process).

Source code in src/crew_dcs/routes/dataset/upload.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def upload_dataset_stage_2_file(
    auth: DomoAuth,
    dataset_id: str,
    upload_id: str,  # must originate from  a stage_1 upload response
    data_file: io.TextIOWrapper | None = None,
    # only necessary if streaming multiple files into the same partition (multi-part upload)
    part_id: int = 2,
    *,  # Make following params keyword-only
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Upload data file to dataset (stage 2 of upload process)."""

    url = f"https://{auth.domo_instance}.domo.com/api/data/v3/datasources/{dataset_id}/uploads/{upload_id}/parts/{part_id}"

    body = data_file

    res = await gd.get_data(
        url=url,
        method="PUT",
        auth=auth,
        content_type="text/csv",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise UploadDataError(stage_num=2, dataset_id=dataset_id, res=res)

    res.upload_id = upload_id
    res.dataset_id = dataset_id
    res.part_id = part_id

    return res

upload_dataset_stage_3 async 

upload_dataset_stage_3(
    auth: DomoAuth,
    dataset_id: str,
    upload_id: str,
    update_method: str = "REPLACE",
    partition_tag: str | None = None,
    is_index: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

commit will close the upload session, upload_id. this request defines how the data will be loaded into Adrenaline, update_method has optional flag for indexing dataset.

Source code in src/crew_dcs/routes/dataset/upload.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def upload_dataset_stage_3(
    auth: DomoAuth,
    dataset_id: str,
    upload_id: str,  # must originate from  a stage_1 upload response
    update_method: str = "REPLACE",  # accepts REPLACE or APPEND
    partition_tag: str | None = None,  # synonymous with data_tag
    is_index: bool = False,  # index after uploading
    *,  # Make following params keyword-only
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """commit will close the upload session, upload_id.  this request defines how the data will be loaded into Adrenaline, update_method
    has optional flag for indexing dataset.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/data/v3/datasources/{dataset_id}/uploads/{upload_id}/commit"

    body = {"index": is_index, "action": update_method}

    if partition_tag:
        body.update(
            {
                "action": "APPEND",
                "dataTag": partition_tag,
                "appendId": "latest" if partition_tag else None,
                "index": is_index,
            }
        )

    res = await gd.get_data(
        auth=auth,
        method="PUT",
        url=url,
        body=body,
        context=context,
    )

    if not res.is_success:
        raise UploadDataError(stage_num=3, dataset_id=dataset_id, res=res)

    res.upload_id = upload_id
    res.dataset_id = dataset_id

    return res

Modules