core

core

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

@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

@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

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

@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_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

@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

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

@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

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

@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