datacenter

datacenter

Datacenter Package

This package provides datacenter management functionality for searching, lineage, and sharing.

Modules:

Name	Description
exceptions	Exception classes for datacenter operations
core	Core datacenter functions, enums, and utilities

DatacenterGetError

DatacenterGetError(
    res=None, message: str | None = None, **kwargs
)

Bases: RouteError

Raised when datacenter retrieval operations fail.

Source code in src/crew_dcs/routes/datacenter/exceptions.py

def __init__(self, res=None, message: str | None = None, **kwargs):
    super().__init__(
        res=res,
        entity_id=(
            getattr(getattr(res, "auth", None), "domo_instance", None)
            if res
            else None
        ),
        message=message or "Datacenter retrieval failed",
        **kwargs,
    )

Lineage_Entity_Type_Enum

Bases: DomoEnumMixin, Enum

Entity types accepted by the lineage API.

Note: These differ from Datacenter_Enum values. The API expects uppercase with underscores (DATA_SOURCE, not DATASET).

SearchDatacenterNoResultsFoundError

SearchDatacenterNoResultsFoundError(
    res=None, message: str | None = None, **kwargs
)

Bases: RouteError

Raised when datacenter search operations return no results.

Source code in src/crew_dcs/routes/datacenter/exceptions.py

def __init__(self, res=None, message: str | None = None, **kwargs):
    super().__init__(
        res=res,
        entity_id=(
            getattr(getattr(res, "auth", None), "domo_instance", None)
            if res
            else None
        ),
        message=message or "No results for query parameters",
        **kwargs,
    )

ShareResourceError

ShareResourceError(
    message: str | None = None,
    domo_instance: str | None = None,
    parent_class: str | None = None,
    function_name: str | None = None,
    res=None,
    **kwargs
)

Bases: RouteError

Raised when resource sharing operations fail.

Source code in src/crew_dcs/routes/datacenter/exceptions.py

def __init__(
    self,
    message: str | None = None,
    domo_instance: str | None = None,
    parent_class: str | None = None,
    function_name: str | None = None,
    res=None,
    **kwargs,
):
    super().__init__(
        message=message or "Resource sharing operation failed",
        domo_instance=domo_instance,
        parent_class=parent_class,
        function_name=function_name,
        res=res,
        **kwargs,
    )

generate_data_center_profile_body

generate_data_center_profile_body(
    *,
    entity_type: (
        str | Datacenter_Enum | list[Datacenter_Enum]
    ),
    filters: list[dict],
    maximum: int | None,
    offset: int = 0
) -> dict

Generate DATA_CENTER profile body for search/v1/query requests.

Source code in src/crew_dcs/routes/datacenter/core.py

def generate_data_center_profile_body(
    *,
    entity_type: str | Datacenter_Enum | list[Datacenter_Enum],
    filters: list[dict],
    maximum: int | None,
    offset: int = 0,
) -> dict:
    """Generate DATA_CENTER profile body for search/v1/query requests."""
    if not isinstance(entity_type, list):
        entity_type = [entity_type]

    entity_type = [dmue.normalize_enum(en) for en in entity_type]

    return {
        "entities": entity_type,
        "count": min(maximum, 100) if maximum else 100,
        "offset": offset,
        "combineResults": False,
        "query": "*",
        "queryProfile": "DATA_CENTER",
        "filters": filters,
        "facetValuesToInclude": None,
        "facetValueLimit": 400,
        "facetValueOffset": 0,
        "includePhonetic": True,
        "useEntities": True,
    }

generate_search_datacenter_account_body

generate_search_datacenter_account_body(
    search_str: str, is_exact_match: bool = True
) -> dict

Generate body for datacenter account search.

Parameters:

Name	Type	Description	Default
search_str	str	String to search for	required
is_exact_match	bool	Whether to require exact match	True

Returns:

Type	Description
dict	Dictionary containing account search body

Source code in src/crew_dcs/routes/datacenter/core.py

def generate_search_datacenter_account_body(
    search_str: str, is_exact_match: bool = True
) -> dict:
    """Generate body for datacenter account search.

    Args:
        search_str: String to search for
        is_exact_match: Whether to require exact match

    Returns:
        Dictionary containing account search body
    """
    return {
        "combineResults": False,
        "query": search_str if is_exact_match else f"*{search_str}*",
        "filters": [],
        "facetValuesToInclude": [
            "DATAPROVIDERNAME",
            "OWNED_BY_ID",
            "VALID",
            "USED",
            "LAST_MODIFIED_DATE",
        ],
        "queryProfile": "GLOBAL",
        "entityList": [["account"]],
        "sort": {"fieldSorts": [{"field": "display_name_sort", "sortOrder": "ASC"}]},
    }

generate_search_datacenter_body

generate_search_datacenter_body(
    search_text: str | None = None,
    entity_type: (
        str | Datacenter_Enum | list[Datacenter_Enum]
    ) = "DATASET",
    additional_filters_ls: list[dict] | None = None,
    combineResults: bool = True,
    limit: int = 100,
    offset: int = 0,
) -> dict

Generate complete body for datacenter search request.

Parameters:

Name	Type	Description	Default
search_text	str | None	Optional search text to filter results	None
entity_type	str | Datacenter_Enum | list[Datacenter_Enum]	Type(s) of entities to search for	'DATASET'
additional_filters_ls	list[dict] | None	Additional filters to apply	None
combineResults	bool	Whether to combine results	True
limit	int	Maximum number of results	100
offset	int	Offset for pagination	0

Returns:

Type	Description
dict	Dictionary containing complete search body

Raises:

Type	Description
ValueError	If entity_type is not a string or Datacenter_Enum

Source code in src/crew_dcs/routes/datacenter/core.py

def generate_search_datacenter_body(
    search_text: str | None = None,
    entity_type: (
        str | Datacenter_Enum | list[Datacenter_Enum]
    ) = "DATASET",  # can accept one entity_type or a list of entity_types
    additional_filters_ls: list[dict] | None = None,
    combineResults: bool = True,
    limit: int = 100,
    offset: int = 0,
) -> dict:
    """Generate complete body for datacenter search request.

    Args:
        search_text: Optional search text to filter results
        entity_type: Type(s) of entities to search for
        additional_filters_ls: Additional filters to apply
        combineResults: Whether to combine results
        limit: Maximum number of results
        offset: Offset for pagination

    Returns:
        Dictionary containing complete search body

    Raises:
        ValueError: If entity_type is not a string or Datacenter_Enum
    """
    filters_ls = (
        [generate_search_datacenter_filter_search_term(search_text)]
        if search_text
        else []
    )

    if not isinstance(entity_type, list):
        entity_type = [entity_type]

    entity_type = [dmue.normalize_enum(en) for en in entity_type]

    if not all(isinstance(en, str) for en in entity_type):
        raise ValueError("entity_type must be a string or Datacenter_Enum")

    if additional_filters_ls:
        if not isinstance(additional_filters_ls, list):
            additional_filters_ls = [additional_filters_ls]

        filters_ls += additional_filters_ls

    return {
        "entities": entity_type,
        "filters": filters_ls or [],
        "combineResults": combineResults,
        "query": "*",
        "count": limit,
        "offset": offset,
    }

generate_search_datacenter_filter

generate_search_datacenter_filter(
    field: str | Datacenter_Filter_Field_Enum,
    value: str | Enum,
    is_not: bool = False,
) -> dict

Generate a filter object for datacenter search.

Parameters:

Name	Type	Description	Default
field	str | Datacenter_Filter_Field_Enum	Field to filter on (string or enum)	required
value	str | Enum	Value to filter for (string or enum)	required
is_not	bool	Whether to negate the filter	False

Returns:

Type	Description
dict	Dictionary containing filter specification

Source code in src/crew_dcs/routes/datacenter/core.py

def generate_search_datacenter_filter(
    field: str | Datacenter_Filter_Field_Enum,  # use Datacenter_Filter_Field_Enum
    value: str | Enum,
    is_not: bool = False,  # to handle exclusion
) -> dict:
    """Generate a filter object for datacenter search.

    Args:
        field: Field to filter on (string or enum)
        value: Value to filter for (string or enum)
        is_not: Whether to negate the filter

    Returns:
        Dictionary containing filter specification
    """
    field = dmue.normalize_enum(field)
    value = dmue.normalize_enum(value)

    return {
        "filterType": "term",
        "field": field,
        "value": value,
        "not": is_not,
    }

generate_search_datacenter_filter_search_term

generate_search_datacenter_filter_search_term(
    search_term: str,
) -> dict

Generate a search term filter for datacenter search.

Parameters:

Name	Type	Description	Default
search_term	str	Text to search for	required

Returns:

Type	Description
dict	Dictionary containing search term filter

Source code in src/crew_dcs/routes/datacenter/core.py

def generate_search_datacenter_filter_search_term(search_term: str) -> dict:
    """Generate a search term filter for datacenter search.

    Args:
        search_term: Text to search for

    Returns:
        Dictionary containing search term filter
    """
    return {"field": "name_sort", "filterType": "wildcard", "query": search_term}

generate_search_datacenter_filters

generate_search_datacenter_filters(
    filter_specs: list[
        tuple[object | None, Callable[[object], dict]]
    ],
) -> list[dict]

Build datacenter filters from declarative (value, builder) specs.

Source code in src/crew_dcs/routes/datacenter/core.py

def generate_search_datacenter_filters(
    filter_specs: list[tuple[object | None, Callable[[object], dict]]],
) -> list[dict]:
    """Build datacenter filters from declarative `(value, builder)` specs."""
    return [
        filter_builder(filter_value)
        for filter_value, filter_builder in filter_specs
        if filter_value is not None
    ]

generate_semantic_model_search_body

generate_semantic_model_search_body(
    *,
    filters: list[dict] | None = None,
    last_updated_bucket: str | None = None,
    maximum: int | None = None,
    offset: int = 0
) -> dict

Generate a SEMANTIC_MODEL profile body for search/v1/query.

Semantic models (a.k.a. data models) are DATASET entities surfaced under the SEMANTIC_MODEL query profile and the "Semantic Model" data-provider facet. This mirrors generate_data_center_profile_body but pins the semantic-model profile + facet filter.

Parameters:

Name	Type	Description	Default
filters	list[dict] | None	Extra filters to append (term/dateBucket/etc.)	None
last_updated_bucket	str | None	Optional last_updated dateBucket value (e.g. "LAST_WEEK"). Omit to return all.	None
maximum	int | None	Maximum number of results (clamped to 100 per request)	None
offset	int	Pagination offset	0

Returns:

Type	Description
dict	Dictionary containing the complete search body

Source code in src/crew_dcs/routes/datacenter/core.py

def generate_semantic_model_search_body(
    *,
    filters: list[dict] | None = None,
    last_updated_bucket: str | None = None,  # e.g. "LAST_WEEK", "LAST_MONTH"
    maximum: int | None = None,
    offset: int = 0,
) -> dict:
    """Generate a SEMANTIC_MODEL profile body for search/v1/query.

    Semantic models (a.k.a. data models) are DATASET entities surfaced under
    the ``SEMANTIC_MODEL`` query profile and the "Semantic Model" data-provider
    facet. This mirrors `generate_data_center_profile_body` but pins the
    semantic-model profile + facet filter.

    Args:
        filters: Extra filters to append (term/dateBucket/etc.)
        last_updated_bucket: Optional ``last_updated`` dateBucket value
            (e.g. "LAST_WEEK"). Omit to return all.
        maximum: Maximum number of results (clamped to 100 per request)
        offset: Pagination offset

    Returns:
        Dictionary containing the complete search body
    """
    filters_ls: list[dict] = [
        {
            "filterType": "term",
            "field": "dataprovidername_facet",
            "value": "Semantic Model",
            "name": "Semantic Model",
            "not": False,
        }
    ]

    if last_updated_bucket:
        filters_ls.append(
            {
                "filterType": "dateBucket",
                "field": "last_updated",
                "value": last_updated_bucket,
            }
        )

    if filters:
        filters_ls += filters

    return {
        "entities": ["DATASET"],
        "count": min(maximum, 100) if maximum else 100,
        "offset": offset,
        "combineResults": False,
        "query": "*",
        "queryProfile": "SEMANTIC_MODEL",
        "filters": filters_ls,
        "facetValuesToInclude": None,
        "facetValueLimit": 400,
        "facetValueOffset": 0,
        "includePhonetic": True,
        "useEntities": True,
    }

get_connectors async

get_connectors(
    auth: DomoAuth,
    search_text: str | None = None,
    additional_filters_ls: list[dict] | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve available connectors from datacenter.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object	required
search_text	str | None	Optional text to filter connectors	None
additional_filters_ls	list[dict] | None	Additional filters to apply	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 connector list

Raises:

Type	Description
SearchDatacenter_NoResultsFound	If no connectors found
Datacenter_GET_Error	If connector retrieval fails

Source code in src/crew_dcs/routes/datacenter/core.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_connectors(
    auth: DomoAuth,
    search_text: str | None = None,
    additional_filters_ls: list[dict] | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve available connectors from datacenter.

    Args:
        auth: Authentication object
        search_text: Optional text to filter connectors
        additional_filters_ls: Additional filters to apply
        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 connector list

    Raises:
        SearchDatacenter_NoResultsFound: If no connectors found
        Datacenter_GET_Error: If connector retrieval fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    additional_filters_ls = additional_filters_ls or []

    body = generate_search_datacenter_body(
        entity_type=Datacenter_Enum.CONNECTOR,
        additional_filters_ls=additional_filters_ls,
        combineResults=True,
    )

    res = await search_datacenter(
        auth=auth,
        body=body,
        return_raw=True,  # Get raw response first to avoid raising errors
        context=context,
    )

    if return_raw:
        return res

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

    if search_text:
        s = [
            r
            for r in res.response
            if search_text.lower() in r.get("label", "").lower()
            or search_text.lower() in r.get("title", "").lower()
        ]

        res.response = s

    return res

get_lineage_upstream async

get_lineage_upstream(
    auth: DomoAuth,
    entity_type: str | Lineage_Entity_Type_Enum,
    entity_id: str,
    return_raw: bool = False,
    max_depth: int | None = None,
    traverse_up: bool = True,
    traverse_down: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Get lineage for a datacenter entity.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object	required
entity_type	str | Lineage_Entity_Type_Enum	Type of entity (use Lineage_Entity_Type_Enum for valid values)	required
entity_id	str	ID of the entity	required
return_raw	bool	Return raw response without processing	False
max_depth	int | None	Maximum depth to traverse. Limits how many levels to include. None = unlimited depth.	None
traverse_up	bool	If True, traverse upstream (parents/dependencies).	True
traverse_down	bool	If True, traverse downstream (children/dependents).	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 lineage data

Raises:

Type	Description
Datacenter_GET_Error	If lineage retrieval fails

Source code in src/crew_dcs/routes/datacenter/core.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_lineage_upstream(
    auth: DomoAuth,
    entity_type: str | Lineage_Entity_Type_Enum,
    entity_id: str,
    return_raw: bool = False,
    max_depth: int | None = None,
    traverse_up: bool = True,
    traverse_down: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get lineage for a datacenter entity.

    Args:
        auth: Authentication object
        entity_type: Type of entity (use Lineage_Entity_Type_Enum for valid values)
        entity_id: ID of the entity
        return_raw: Return raw response without processing
        max_depth: Maximum depth to traverse. Limits how many levels
                  to include. None = unlimited depth.
        traverse_up: If True, traverse upstream (parents/dependencies).
        traverse_down: If True, traverse downstream (children/dependents).
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing lineage data

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

    # Normalize entity_type to string value
    entity_type = dmue.normalize_enum(entity_type)

    url = f"https://{auth.domo_instance}.domo.com/api/data/v1/lineage/{entity_type}/{entity_id}"

    params = {
        "traverseUp": "true" if traverse_up else "false",
        "traverseDown": "true" if traverse_down else "false",
    }

    # Add maxDepth parameter if specified
    if max_depth is not None:
        params.update({"maxDepth": str(max_depth)})

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

    if return_raw:
        return res

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

    return res

search_datacenter async

search_datacenter(
    auth: DomoAuth,
    maximum: int | None = None,
    body: dict | None = None,
    search_text: str | None = None,
    entity_type: str | list = "dataset",
    additional_filters_ls: list | None = None,
    arr_fn: Callable | None = None,
    debug_loop: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Search across datacenter entities.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object	required
maximum	int | None	Maximum number of results to return	None
body	dict | None	Pre-built search body (optional)	None
search_text	str | None	Text to search for	None
entity_type	str | list	Type(s) of entities to search	'dataset'
additional_filters_ls	list | None	Additional filters to apply	None
arr_fn	Callable | None	Function to extract array from response	None
debug_loop	bool	Enable loop debugging	False
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 search results

Raises:

Type	Description
SearchDatacenter_NoResultsFound	If search returns no results
Datacenter_GET_Error	If search operation fails

Source code in src/crew_dcs/routes/datacenter/core.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def search_datacenter(
    auth: DomoAuth,
    maximum: int | None = None,
    body: (
        dict | None
    ) = None,  # either pass a body or generate a body in the function using search_text, entity_type, and additional_filters parameters
    search_text: str | None = None,
    entity_type: str | list = "dataset",  # can accept one value or a list of values
    additional_filters_ls: list | None = None,
    arr_fn: Callable | None = None,
    debug_loop: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Search across datacenter entities.

    Args:
        auth: Authentication object
        maximum: Maximum number of results to return
        body: Pre-built search body (optional)
        search_text: Text to search for
        entity_type: Type(s) of entities to search
        additional_filters_ls: Additional filters to apply
        arr_fn: Function to extract array from response
        debug_loop: Enable loop debugging
        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 search results

    Raises:
        SearchDatacenter_NoResultsFound: If search returns no results
        Datacenter_GET_Error: If search operation fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    limit = 100  # api enforced limit

    if not body:
        body = generate_search_datacenter_body(
            entity_type=entity_type,
            additional_filters_ls=additional_filters_ls,
            search_text=search_text,
            combineResults=False,
            limit=limit,
        )

    if not arr_fn:

        def arr_fn(res):
            return res.response.get("searchObjects")

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

    res = await gd.looper(
        auth=auth,
        url=url,
        loop_until_end=bool(not maximum),
        body=body,
        offset_params_in_body=True,
        offset_params={"offset": "offset", "limit": "count"},
        arr_fn=arr_fn,
        method="POST",
        maximum=maximum,
        limit=limit,
        context=context,
        debug_loop=debug_loop,
        return_raw=return_raw,
    )

    if return_raw:
        return res

    if res.is_success and len(res.response) == 0:
        raise SearchDatacenterNoResultsFoundError(
            res=res, message="no results for query parameters"
        )

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

    return res

search_semantic_models async

search_semantic_models(
    auth: DomoAuth,
    maximum: int | None = None,
    last_updated_bucket: str | None = None,
    additional_filters_ls: list[dict] | None = None,
    body: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Search Domo semantic models (data models) via the SEMANTIC_MODEL profile.

A semantic model is a DATASET-entity datasource of type data-model; this searches them through the SEMANTIC_MODEL query profile + the "Semantic Model" data-provider facet.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object	required
maximum	int | None	Maximum number of results to return	None
last_updated_bucket	str | None	Optional last_updated dateBucket (e.g. "LAST_WEEK")	None
additional_filters_ls	list[dict] | None	Extra filters to append to the body	None
body	dict | None	Pre-built search body (overrides the generated one)	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 containing the matching semantic-model search objects

Raises:

Type	Description
SearchDatacenterNoResultsFoundError	If search returns no results
DatacenterGetError	If the search operation fails

Source code in src/crew_dcs/routes/datacenter/core.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def search_semantic_models(
    auth: DomoAuth,
    maximum: int | None = None,
    last_updated_bucket: str | None = None,
    additional_filters_ls: list[dict] | None = None,
    body: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Search Domo semantic models (data models) via the SEMANTIC_MODEL profile.

    A semantic model is a DATASET-entity datasource of type ``data-model``;
    this searches them through the ``SEMANTIC_MODEL`` query profile + the
    "Semantic Model" data-provider facet.

    Args:
        auth: Authentication object
        maximum: Maximum number of results to return
        last_updated_bucket: Optional ``last_updated`` dateBucket (e.g. "LAST_WEEK")
        additional_filters_ls: Extra filters to append to the body
        body: Pre-built search body (overrides the generated one)
        return_raw: Return raw response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData containing the matching semantic-model search objects

    Raises:
        SearchDatacenterNoResultsFoundError: If search returns no results
        DatacenterGetError: If the search operation fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    if not body:
        body = generate_semantic_model_search_body(
            filters=additional_filters_ls,
            last_updated_bucket=last_updated_bucket,
            maximum=maximum,
        )

    return await search_datacenter(
        auth=auth,
        body=body,
        maximum=maximum,
        return_raw=return_raw,
        context=context,
    )

share_resource async

share_resource(
    auth: DomoAuth,
    resource_ids: list | str | int,
    resource_type: ShareResource_Enum,
    group_ids: list | str | int | None = None,
    user_ids: list | str | int | None = None,
    message: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Share a page or card with users or groups.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object	required
resource_ids	list | str | int	ID(s) of resources to share	required
resource_type	ShareResource_Enum	Type of resource (ShareResource_Enum enum)	required
group_ids	list | str | int | None	ID(s) of groups to share with	None
user_ids	list | str | int | None	ID(s) of users to share with	None
message	str | None	Optional message to include in notification email	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 with success message

Raises:

Type	Description
ShareResource_Error	If sharing operation fails

Example body format

{ "resources": [ { "type": "page", "id": {page_id} } ], "recipients": [ { "type": "group", "id": "{group_id}" } ], "message": "I thought you might find this page interesting." }

Source code in src/crew_dcs/routes/datacenter/core.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def share_resource(
    auth: DomoAuth,
    resource_ids: list | str | int,
    resource_type: ShareResource_Enum,
    group_ids: list | str | int | None = None,
    user_ids: list | str | int | None = None,
    message: str | None = None,  # email to user
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Share a page or card with users or groups.

    Args:
        auth: Authentication object
        resource_ids: ID(s) of resources to share
        resource_type: Type of resource (ShareResource_Enum enum)
        group_ids: ID(s) of groups to share with
        user_ids: ID(s) of users to share with
        message: Optional message to include in notification email
        return_raw: Return raw response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object with success message

    Raises:
        ShareResource_Error: If sharing operation fails

    Example body format:
        {
            "resources": [
                {
                    "type": "page",
                    "id": {page_id}
                }
            ],
            "recipients": [
                {
                    "type": "group",
                    "id": "{group_id}"
                }
            ],
            "message": "I thought you might find this page interesting."
        }
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    resource_ids = resource_ids if isinstance(resource_ids, list) else [resource_ids]
    if group_ids:
        group_ids = (
            group_ids and group_ids if isinstance(group_ids, list) else [group_ids]
        )

    if user_ids:
        user_ids = user_ids if isinstance(user_ids, list) else [user_ids]

    url = f"https://{auth.domo_instance}.domo.com/api/content/v1/share?sendEmail=false"

    recipient_ls = []

    if group_ids:
        for gid in group_ids:
            recipient_ls.append({"type": "group", "id": str(gid)})

    if user_ids:
        for uid in user_ids:
            recipient_ls.append({"type": "user", "id": str(uid)})

    resource_ls = [
        {"type": resource_type.value, "id": str(rid)} for rid in resource_ids
    ]

    body = {
        "resources": resource_ls,
        "recipients": recipient_ls,
        "message": message,
    }

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

    if return_raw:
        return res

    if not res.is_success:
        raise ShareResourceError(
            message=res.response,
            domo_instance=auth.domo_instance,
            parent_class=context.parent_class,
            function_name="share_resource",
            res=res,
        )

    if res.is_success:
        res.response = f"{resource_type.value} {','.join([resource['id'] for resource in resource_ls])} successfully shared with {', '.join([recipient['id'] for recipient in recipient_ls])}"

    return res

Modules

• core
• exceptions