datastores

datastores

create_datastore async

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

Create a new datastore.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
datastore_name	str | None	Name for the new datastore (shows as appName in UI)	None
return_raw	bool	Return raw API 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 created datastore information

Raises:

Type	Description
AppDb_CRUD_Error	If datastore creation fails

Source code in src/crew_dcs/routes/appdb/datastores.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_datastore(
    auth: DomoAuth,
    datastore_name: str | None = None,  # in UI shows up as appName
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a new datastore.

    Args:
        auth: Authentication object containing credentials and instance info
        datastore_name: Name for the new datastore (shows as appName in UI)
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing created datastore information

    Raises:
        AppDb_CRUD_Error: If datastore creation fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

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

    body = {"name": datastore_name}

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

    if return_raw:
        return res

    if not res.is_success:
        raise AppDb_CRUD_Error(operation="create", res=res)

    return res

delete_datastore async

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

Delete a datastore by ID.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
datastore_id	str	Unique identifier for the datastore	required
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	RouteContext for request configuration	None
**context_kwargs		Additional context parameters	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing delete result

Raises:

Type	Description
AppDb_CRUD_Error	If datastore deletion fails

Source code in src/crew_dcs/routes/appdb/datastores.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def delete_datastore(
    auth: DomoAuth,
    datastore_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Delete a datastore by ID.

    Args:
        auth: Authentication object containing credentials and instance info
        datastore_id: Unique identifier for the datastore
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters

    Returns:
        ResponseGetData object containing delete result

    Raises:
        AppDb_CRUD_Error: If datastore deletion fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

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

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

    if return_raw:
        return res

    if not res.is_success:
        raise AppDb_CRUD_Error(operation="delete", appdb_id=datastore_id, res=res)

    return res

get_collections_from_datastore async

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

Get collections from a specific datastore.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
datastore_id	str	Unique identifier for the datastore	required
return_raw	bool	Return raw API 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 collections information

Raises:

Type	Description
AppDb_GET_Error	If collections retrieval fails

Source code in src/crew_dcs/routes/appdb/datastores.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_collections_from_datastore(
    auth: DomoAuth,
    datastore_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get collections from a specific datastore.

    Args:
        auth: Authentication object containing credentials and instance info
        datastore_id: Unique identifier for the datastore
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing collections information

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

    url = f"https://{auth.domo_instance}.domo.com/api/datastores/v1/{datastore_id}/collections"

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

    if return_raw:
        return res

    if not res.is_success:
        raise AppDb_GET_Error(appdb_id=datastore_id, res=res)

    return res

get_datastore_by_id async

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

Retrieve a specific datastore by ID.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
datastore_id	str	Unique identifier for the datastore	required
return_raw	bool	Return raw API 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 datastore information

Raises:

Type	Description
AppDb_GET_Error	If datastore retrieval fails
SearchAppDbNotFoundError	If datastore with specified ID doesn't exist

Source code in src/crew_dcs/routes/appdb/datastores.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_datastore_by_id(
    auth: DomoAuth,
    datastore_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve a specific datastore by ID.

    Args:
        auth: Authentication object containing credentials and instance info
        datastore_id: Unique identifier for the datastore
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing datastore information

    Raises:
        AppDb_GET_Error: If datastore retrieval fails
        SearchAppDbNotFoundError: If datastore with specified ID doesn't exist
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

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

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

    if return_raw:
        return res

    if not res.is_success:
        if res.status == 404:
            raise SearchAppDbNotFoundError(
                search_criteria=f"datastore_id: {datastore_id}",
                res=res,
            )
        raise AppDb_GET_Error(appdb_id=datastore_id, res=res)

    return res

get_datastore_cards async

get_datastore_cards(
    auth: DomoAuth,
    datastore_ids: list[str],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Get cards associated with one or more datastore IDs.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
datastore_ids	list[str]	List of datastore UUIDs to look up cards for	required
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	RouteContext for request configuration	None
**context_kwargs		Additional context parameters	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing datastore card mappings

Raises:

Type	Description
AppDb_GET_Error	If datastore cards lookup fails

Source code in src/crew_dcs/routes/appdb/datastores.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_datastore_cards(
    auth: DomoAuth,
    datastore_ids: list[str],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get cards associated with one or more datastore IDs.

    Args:
        auth: Authentication object containing credentials and instance info
        datastore_ids: List of datastore UUIDs to look up cards for
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters

    Returns:
        ResponseGetData object containing datastore card mappings

    Raises:
        AppDb_GET_Error: If datastore cards lookup fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/domoapps/apps/v2/card"

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

    if return_raw:
        return res

    if not res.is_success:
        raise AppDb_GET_Error(message="unable to retrieve datastore cards", res=res)

    return res

get_datastores async

get_datastores(
    auth: DomoAuth,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve all datastores.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
context	RouteContext | None	Optional RouteContext for request configuration	None
return_raw	bool	Return raw API response without processing	False
**context_kwargs		Additional context parameters (session, debug_api, etc.)	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing datastores information

Raises:

Type	Description
AppDb_GET_Error	If datastores retrieval fails

Source code in src/crew_dcs/routes/appdb/datastores.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_datastores(
    auth: DomoAuth,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve all datastores.

    Args:
        auth: Authentication object containing credentials and instance info
        context: Optional RouteContext for request configuration
        return_raw: Return raw API response without processing
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing datastores information

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

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

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

    if return_raw:
        return res

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

    return res

update_datastore async

update_datastore(
    auth: DomoAuth,
    datastore_id: str,
    body: dict,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update datastore properties by ID.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
datastore_id	str	Unique identifier for the datastore	required
body	dict	Datastore update payload	required
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	RouteContext for request configuration	None
**context_kwargs		Additional context parameters	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing update result

Raises:

Type	Description
AppDb_CRUD_Error	If datastore update fails

Source code in src/crew_dcs/routes/appdb/datastores.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def update_datastore(
    auth: DomoAuth,
    datastore_id: str,
    body: dict,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update datastore properties by ID.

    Args:
        auth: Authentication object containing credentials and instance info
        datastore_id: Unique identifier for the datastore
        body: Datastore update payload
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters

    Returns:
        ResponseGetData object containing update result

    Raises:
        AppDb_CRUD_Error: If datastore update fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

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

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

    if return_raw:
        return res

    if not res.is_success:
        raise AppDb_CRUD_Error(operation="update", appdb_id=datastore_id, res=res)

    return res