documents

documents

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

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