appdb

appdb ¶

AppDb Route Functions

This module provides functions for managing Domo AppDb datastores, collections, and documents. The functionality is organized into logical submodules.

Submodules

exceptions: Exception classes for AppDb operations datastores: Datastore management functions collections: Collection management functions documents: Document management functions

Exception Classes

AppDb_GET_Error: Raised when AppDb retrieval operations fail SearchAppDbNotFoundError: Raised when AppDb search returns no results AppDb_CRUD_Error: Raised when AppDb create/update/delete operations fail

Datastore Functions

get_datastores: Retrieve all datastores get_datastore_by_id: Retrieve a specific datastore by ID get_collections_from_datastore: Get collections from a specific datastore create_datastore: Create a new datastore

Collection Functions

create_collection: Create a new collection in a datastore get_collections: Retrieve all collections get_collection_by_id: Retrieve a specific collection by ID modify_collection_permissions: Modify collection permissions

Document Functions

get_documents_from_collection: Get documents from a collection get_collection_document_by_id: Get a specific document by ID create_document: Create a new document in a collection update_document: Update an existing document

Enums

Collection_Permission_Enum: Permissions for collection access

AppDb_CRUD_Error ¶

AppDb_CRUD_Error(
    operation: str,
    appdb_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs
)

Bases: RouteError

Raised when AppDb create, update, or delete operations fail.

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

def __init__(
    self,
    operation: str,
    appdb_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs,
):
    super().__init__(
        message=message or f"AppDb {operation} operation failed",
        entity_id=appdb_id,
        res=res,
        **kwargs,
    )

AppDb_GET_Error ¶

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

Bases: RouteError

Raised when AppDb retrieval operations fail.

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

def __init__(
    self,
    appdb_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs,
):
    super().__init__(
        message=message or "AppDb retrieval failed",
        entity_id=appdb_id,
        res=res,
        **kwargs,
    )

SearchAppDbNotFoundError ¶

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

Bases: RouteError

Raised when AppDb search operations return no results.

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

def __init__(
    self,
    search_criteria: str,
    message: str | None = None,
    res=None,
    **kwargs,
):
    super().__init__(
        message=message or f"No AppDb items found matching: {search_criteria}",
        res=res,
        **kwargs,
    )

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

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

create_document `async` ¶

create_document(
    auth: DomoAuth,
    collection_id: str,
    content: dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create a new document in 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
`content`	`dict[str, Any]`	Document content to create	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 created document information

Raises:

Type	Description
`AppDb_CRUD_Error`	If document creation fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_document(
    auth: DomoAuth,
    collection_id: str,
    content: dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a new document in a collection.

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

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

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

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

    if return_raw:
        return res

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

    return res

create_documents_bulk `async` ¶

create_documents_bulk(
    auth: DomoAuth,
    collection_id: str,
    documents: list[dict[str, Any]],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create multiple documents in a collection in a single request.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object containing credentials and instance info	required
`collection_id`	`str`	Unique identifier for the collection	required
`documents`	`list[dict[str, Any]]`	List of document content dicts to create	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 created documents information

Raises:

Type	Description
`AppDb_CRUD_Error`	If bulk creation fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_documents_bulk(
    auth: DomoAuth,
    collection_id: str,
    documents: list[dict[str, Any]],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create multiple documents in a collection in a single request.

    Args:
        auth: Authentication object containing credentials and instance info
        collection_id: Unique identifier for the collection
        documents: List of document content dicts to create
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters

    Returns:
        ResponseGetData object containing created documents information

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

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

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

    if return_raw:
        return res

    if not res.is_success:
        raise AppDb_CRUD_Error(operation="create_bulk", appdb_id=collection_id, 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

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

delete_document `async` ¶

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

Delete a single document from 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
`document_id`	`str`	Unique identifier for the document to delete	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 document deletion fails
`SearchAppDbNotFoundError`	If document with specified ID doesn't exist

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

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

    Args:
        auth: Authentication object containing credentials and instance info
        collection_id: Unique identifier for the collection
        document_id: Unique identifier for the document to delete
        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 document deletion fails
        SearchAppDbNotFoundError: If document with specified ID doesn't exist
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/datastores/v2/collections/{collection_id}/documents/{document_id}"

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

    if return_raw:
        return res

    if not res.is_success:
        if res.status == 404:
            raise SearchAppDbNotFoundError(
                search_criteria=f"document_id: {document_id} in collection: {collection_id}",
                res=res,
            )
        raise AppDb_CRUD_Error(operation="delete", appdb_id=document_id, res=res)

    return res

delete_documents_bulk `async` ¶

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

Delete multiple documents from a collection by their IDs.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object containing credentials and instance info	required
`collection_id`	`str`	Unique identifier for the collection	required
`document_ids`	`list[str]`	List of document UUIDs to delete	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 deletion result

Raises:

Type	Description
`AppDb_CRUD_Error`	If bulk deletion fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def delete_documents_bulk(
    auth: DomoAuth,
    collection_id: str,
    document_ids: list[str],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Delete multiple documents from a collection by their IDs.

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

    Returns:
        ResponseGetData object containing deletion result

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

    url = f"https://{auth.domo_instance}.domo.com/api/datastores/v2/collections/{collection_id}/documents/bulk"

    res = await gd.get_data(
        auth=auth,
        method="DELETE",
        url=url,
        params={"ids": ",".join(document_ids)},
        context=context,
    )

    if return_raw:
        return res

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

    return res

download_documents `async` ¶

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

Download all documents from 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 the downloaded documents

Raises:

Type	Description
`AppDb_GET_Error`	If download fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def download_documents(
    auth: DomoAuth,
    collection_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Download all documents from 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 the downloaded documents

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

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

    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=f"unable to download documents from collection {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_document_by_id `async` ¶

get_collection_document_by_id(
    auth: DomoAuth,
    collection_id: str,
    document_id: str,
    query: dict[str, Any] | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Get a specific document by ID from 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
`document_id`	`str`	Unique identifier for the document	required
`query`	`dict[str, Any] \| None`	Optional query parameters	`None`
`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 document information

Raises:

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

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_collection_document_by_id(
    auth: DomoAuth,
    collection_id: str,
    document_id: str,
    query: dict[str, Any] | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get a specific document by ID from a collection.

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

    Raises:
        AppDb_GET_Error: If document retrieval fails
        SearchAppDbNotFoundError: If document 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}/documents/{document_id}"

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

    if return_raw:
        return res

    if not res.is_success:
        if res.status == 404:
            raise SearchAppDbNotFoundError(
                search_criteria=f"document_id: {document_id} in collection: {collection_id}",
                res=res,
            )
        raise AppDb_GET_Error(appdb_id=document_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

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

get_documents_from_collection `async` ¶

get_documents_from_collection(
    auth: DomoAuth,
    collection_id: str,
    query: dict[str, Any] | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Get documents from 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
`query`	`dict[str, Any] \| None`	Optional query parameters for document filtering	`None`
`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 documents information

Raises:

Type	Description
`AppDb_GET_Error`	If documents retrieval fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_documents_from_collection(
    auth: DomoAuth,
    collection_id: str,
    query: dict[str, Any] | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get documents from a collection.

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

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

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

    query = query or {}

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

    if return_raw:
        return res

    if not res.is_success:
        raise AppDb_GET_Error(
            appdb_id=collection_id,
            message=f"unable to query documents in collection - {collection_id}",
            res=res,
        )

    return res

import_documents `async` ¶

import_documents(
    auth: DomoAuth,
    collection_id: str,
    documents: list[dict[str, Any]],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Import documents into 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
`documents`	`list[dict[str, Any]]`	List of document dicts to import	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 import result

Raises:

Type	Description
`AppDb_CRUD_Error`	If import fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def import_documents(
    auth: DomoAuth,
    collection_id: str,
    documents: list[dict[str, Any]],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Import documents into a collection.

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

    Returns:
        ResponseGetData object containing import result

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

    url = f"https://{auth.domo_instance}.domo.com/api/datastores/v2/collections/{collection_id}/documents/import"

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

    if return_raw:
        return res

    if not res.is_success:
        raise AppDb_CRUD_Error(operation="import", appdb_id=collection_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

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

update_document `async` ¶

update_document(
    auth: DomoAuth,
    collection_id: str,
    document_id: str,
    content: dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update an existing document in 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
`document_id`	`str`	Unique identifier for the document to update	required
`content`	`dict[str, Any]`	Updated document content	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 updated document information

Raises:

Type	Description
`AppDb_CRUD_Error`	If document update fails
`SearchAppDbNotFoundError`	If document with specified ID doesn't exist

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def update_document(
    auth: DomoAuth,
    collection_id: str,
    document_id: str,
    content: dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update an existing document in a collection.

    Args:
        auth: Authentication object containing credentials and instance info
        collection_id: Unique identifier for the collection
        document_id: Unique identifier for the document to update
        content: Updated document content
        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 updated document information

    Raises:
        AppDb_CRUD_Error: If document update fails
        SearchAppDbNotFoundError: If document with specified ID doesn't exist
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/datastores/v2/collections/{collection_id}/documents/{document_id}"

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

    if return_raw:
        return res

    if not res.is_success:
        if res.status == 404:
            raise SearchAppDbNotFoundError(
                search_criteria=f"document_id: {document_id} in collection: {collection_id}",
                res=res,
            )
        raise AppDb_CRUD_Error(operation="update", appdb_id=document_id, res=res)

    return res

update_documents_bulk `async` ¶

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

Operator bulk update documents matching criteria in 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
`body`	`dict[str, Any]`	Update payload (filter criteria + update operations)	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 bulk update fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def update_documents_bulk(
    auth: DomoAuth,
    collection_id: str,
    body: dict[str, Any],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Operator bulk update documents matching criteria in a collection.

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

    url = f"https://{auth.domo_instance}.domo.com/api/datastores/v2/collections/{collection_id}/documents/update"

    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_bulk", appdb_id=collection_id, res=res)

    return res

upsert_documents_bulk `async` ¶

upsert_documents_bulk(
    auth: DomoAuth,
    collection_id: str,
    documents: list[dict[str, Any]],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Upsert multiple documents in a collection in a single request.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object containing credentials and instance info	required
`collection_id`	`str`	Unique identifier for the collection	required
`documents`	`list[dict[str, Any]]`	List of document dicts to upsert (must include id for existing docs)	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 upserted documents information

Raises:

Type	Description
`AppDb_CRUD_Error`	If bulk upsert fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def upsert_documents_bulk(
    auth: DomoAuth,
    collection_id: str,
    documents: list[dict[str, Any]],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Upsert multiple documents in a collection in a single request.

    Args:
        auth: Authentication object containing credentials and instance info
        collection_id: Unique identifier for the collection
        documents: List of document dicts to upsert (must include id for existing docs)
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters

    Returns:
        ResponseGetData object containing upserted documents information

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

    url = f"https://{auth.domo_instance}.domo.com/api/datastores/v2/collections/{collection_id}/documents/bulk"

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

    if return_raw:
        return res

    if not res.is_success:
        raise AppDb_CRUD_Error(operation="upsert_bulk", 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

appdb

appdb ¶

AppDb_CRUD_Error ¶

AppDb_GET_Error ¶

SearchAppDbNotFoundError ¶

check_user_edit_permission async ¶

create_collection async ¶

create_collection_and_datastore async ¶

create_datastore async ¶

create_document async ¶

create_documents_bulk async ¶

delete_collection async ¶

delete_datastore async ¶

delete_document async ¶

delete_documents_bulk async ¶

download_documents async ¶

get_collection_by_id async ¶

get_collection_document_by_id async ¶

get_collection_permissions async ¶

get_collections async ¶

get_collections_from_datastore async ¶

get_datastore_by_id async ¶

get_datastore_cards async ¶

get_datastores async ¶

get_documents_from_collection async ¶

import_documents async ¶

modify_collection_permissions async ¶

remove_collection_access async ¶

search_collections async ¶

set_single_collection_permission async ¶

update_collection async ¶

update_datastore async ¶

update_document async ¶

update_documents_bulk async ¶

upsert_documents_bulk async ¶

validate_collection async ¶

Modules¶

check_user_edit_permission `async` ¶

create_collection `async` ¶

create_collection_and_datastore `async` ¶

create_datastore `async` ¶

create_document `async` ¶

create_documents_bulk `async` ¶

delete_collection `async` ¶

delete_datastore `async` ¶

delete_document `async` ¶

delete_documents_bulk `async` ¶

download_documents `async` ¶

get_collection_by_id `async` ¶

get_collection_document_by_id `async` ¶

get_collection_permissions `async` ¶

get_collections `async` ¶

get_collections_from_datastore `async` ¶

get_datastore_by_id `async` ¶

get_datastore_cards `async` ¶

get_datastores `async` ¶

get_documents_from_collection `async` ¶

import_documents `async` ¶

modify_collection_permissions `async` ¶

remove_collection_access `async` ¶

search_collections `async` ¶

set_single_collection_permission `async` ¶

update_collection `async` ¶

update_datastore `async` ¶

update_document `async` ¶

update_documents_bulk `async` ¶

upsert_documents_bulk `async` ¶

validate_collection `async` ¶