collections

collections

check_user_edit_permission async

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

Check edit permission for authenticated user on a collection.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
collection_id	str	Unique identifier for the collection	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 permission check result

Raises:

Type	Description
AppDb_GET_Error	If check fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def check_user_edit_permission(
    auth: DomoAuth,
    collection_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Check edit permission for authenticated user on a collection.

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

    Returns:
        ResponseGetData object containing permission check result

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

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

    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=collection_id,
            message="unable to check user edit permission",
            res=res,
        )

    return res

create_collection async

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

Create a new collection in a datastore.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
datastore_id	str	Datastore ID where collection will be created	required
collection_name	str	Name for the new collection	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 created collection information

Raises:

Type	Description
AppDb_CRUD_Error	If collection creation fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_collection(
    auth: DomoAuth,
    datastore_id: str,  # collections must be created inside a datastore which will show as the associated app_name
    collection_name: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a new collection in a datastore.

    Args:
        auth: Authentication object containing credentials and instance info
        datastore_id: Datastore ID where collection will be created
        collection_name: Name for the new collection
        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 created collection information

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

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

    body = {"name": collection_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", appdb_id=datastore_id, res=res)

    return res

create_collection_and_datastore async

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

Create a collection and datastore in one request.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
body	dict	Request payload accepted by /datastores/v1/collections	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 create result

Raises:

Type	Description
AppDb_CRUD_Error	If creation fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_collection_and_datastore(
    auth: DomoAuth,
    body: dict,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a collection and datastore in one request.

    Args:
        auth: Authentication object containing credentials and instance info
        body: Request payload accepted by /datastores/v1/collections
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters

    Returns:
        ResponseGetData object containing create result

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

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

    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_collection async

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

Delete a collection by ID.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
collection_id	str	Unique identifier for the collection	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 collection deletion fails

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

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

    Args:
        auth: Authentication object containing credentials and instance info
        collection_id: Unique identifier for the collection
        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 collection deletion fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/datastores/v1/collections/{collection_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=collection_id, res=res)

    return res

get_collection_by_id async

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

Retrieve a specific collection by ID.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
collection_id	str	Unique identifier for the collection	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 collection information

Raises:

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

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

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

    Args:
        auth: Authentication object containing credentials and instance info
        collection_id: Unique identifier for the collection
        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 collection information

    Raises:
        AppDb_GET_Error: If collection retrieval fails
        SearchAppDbNotFoundError: If collection 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/collections/{collection_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"collection_id: {collection_id}",
                res=res,
            )
        raise AppDb_GET_Error(appdb_id=collection_id, res=res)

    return res

get_collection_permissions async

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

Retrieve permissions for a collection.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
collection_id	str	Unique identifier for the collection	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 permission details

Raises:

Type	Description
AppDb_GET_Error	If retrieval fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_collection_permissions(
    auth: DomoAuth,
    collection_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve permissions for a collection.

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

    Returns:
        ResponseGetData object containing permission details

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

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

    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=collection_id,
            message="unable to get collection permissions",
            res=res,
        )

    return res

get_collections async

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

Retrieve all collections, optionally filtered by datastore ID.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
datastore_id	str | None	Optional datastore ID to filter collections	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 collections information

Raises:

Type	Description
AppDb_GET_Error	If collections retrieval fails

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

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

    Args:
        auth: Authentication object containing credentials and instance info
        datastore_id: Optional datastore ID to filter collections
        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/collections/"

    res = await gd.get_data(
        auth=auth,
        method="GET",
        url=url,
        params={"datastoreId": datastore_id},
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success and res.status == 400:
        raise AppDb_GET_Error(
            appdb_id=datastore_id,
            message=f"invalid datastoreId - {datastore_id} or ensure it's shared with authenticated user",
            res=res,
        )

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

    return res

modify_collection_permissions async

modify_collection_permissions(
    auth: DomoAuth,
    collection_id: str,
    user_id: str | None = None,
    group_id: str | None = None,
    permission: Collection_Permission_Enum = READ_CONTENT,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Modify collection permissions for users or groups.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
collection_id	str	Unique identifier for the collection	required
user_id	str | None	Optional user ID to grant permissions to	None
group_id	str | None	Optional group ID to grant permissions to	None
permission	Collection_Permission_Enum	Permission level to grant	READ_CONTENT
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 permission modification result

Raises:

Type	Description
AppDb_CRUD_Error	If permission modification fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def modify_collection_permissions(
    auth: DomoAuth,
    collection_id: str,
    user_id: str | None = None,
    group_id: str | None = None,
    permission: Collection_Permission_Enum = Collection_Permission_Enum.READ_CONTENT,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Modify collection permissions for users or groups.

    Args:
        auth: Authentication object containing credentials and instance info
        collection_id: Unique identifier for the collection
        user_id: Optional user ID to grant permissions to
        group_id: Optional group ID to grant permissions to
        permission: Permission level to grant
        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 permission modification result

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

    url = f"https://{auth.domo_instance}.domo.com/api/datastores/v1/collections/{collection_id}/permission/{'USER' if user_id else 'GROUP'}/{user_id or group_id}"

    params = {
        "overwrite": False,
        "permissions": dmue.normalize_enum(permission),
    }

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

    if return_raw:
        return res

    if not res.is_success:
        raise AppDb_CRUD_Error(
            operation="modify permissions",
            appdb_id=collection_id,
            message=f"unable to set permissions for {user_id or group_id} to {dmue.normalize_enum(permission)} in collection {collection_id}",
            res=res,
        )

    res.response = f"set permissions for {user_id or group_id} to {dmue.normalize_enum(permission)} in collection {collection_id}"

    return res

remove_collection_access async

remove_collection_access(
    auth: DomoAuth,
    collection_id: str,
    entity_type: str,
    entity_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Remove collection access for an entity.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
collection_id	str	Unique identifier for the collection	required
entity_type	str	Entity type (for example USER or GROUP)	required
entity_id	str	Entity identifier	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 permission removal fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def remove_collection_access(
    auth: DomoAuth,
    collection_id: str,
    entity_type: str,
    entity_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Remove collection access for an entity.

    Args:
        auth: Authentication object containing credentials and instance info
        collection_id: Unique identifier for the collection
        entity_type: Entity type (for example USER or GROUP)
        entity_id: Entity identifier
        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 permission removal fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = (
        f"https://{auth.domo_instance}.domo.com/api/datastores/v1/collections/"
        f"{collection_id}/permission/{entity_type}/{entity_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="remove access", appdb_id=collection_id, res=res
        )

    return res

search_collections async

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

Search collections via AppDB query endpoint.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
body	dict	Search payload for /datastores/v1/collections/query	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 search results

Raises:

Type	Description
AppDb_GET_Error	If search fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def search_collections(
    auth: DomoAuth,
    body: dict,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Search collections via AppDB query endpoint.

    Args:
        auth: Authentication object containing credentials and instance info
        body: Search payload for /datastores/v1/collections/query
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters

    Returns:
        ResponseGetData object containing search results

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

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

    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_GET_Error(message="unable to search collections", res=res)

    return res

set_single_collection_permission async

set_single_collection_permission(
    auth: DomoAuth,
    collection_id: str,
    source_type: str,
    source_id: str,
    permission: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Set a single permission tuple for a collection.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
collection_id	str	Unique identifier for the collection	required
source_type	str	Permission source type (for example USER or GROUP)	required
source_id	str	Source identifier	required
permission	str	Permission string	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 permission update fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def set_single_collection_permission(
    auth: DomoAuth,
    collection_id: str,
    source_type: str,
    source_id: str,
    permission: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Set a single permission tuple for a collection.

    Args:
        auth: Authentication object containing credentials and instance info
        collection_id: Unique identifier for the collection
        source_type: Permission source type (for example USER or GROUP)
        source_id: Source identifier
        permission: Permission string
        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 permission update fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = (
        f"https://{auth.domo_instance}.domo.com/api/datastores/v1/collections/"
        f"{collection_id}/permission/{source_type}/{source_id}/{permission}"
    )

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

    if return_raw:
        return res

    if not res.is_success:
        raise AppDb_CRUD_Error(
            operation="set permission", appdb_id=collection_id, res=res
        )

    return res

update_collection async

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

Update collection properties by ID.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
collection_id	str	Unique identifier for the collection	required
body	dict	Collection 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 collection update fails

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

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

    Args:
        auth: Authentication object containing credentials and instance info
        collection_id: Unique identifier for the collection
        body: Collection 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 collection update fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/datastores/v1/collections/{collection_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=collection_id, res=res)

    return res

validate_collection async

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

Validate a collection definition before creation.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing credentials and instance info	required
body	dict	Collection definition to validate	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 validation result

Raises:

Type	Description
AppDb_CRUD_Error	If validation request fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def validate_collection(
    auth: DomoAuth,
    body: dict,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Validate a collection definition before creation.

    Args:
        auth: Authentication object containing credentials and instance info
        body: Collection definition to validate
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters

    Returns:
        ResponseGetData object containing validation result

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

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

    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="validate", appdb_id=None, res=res)

    return res