account

account ¶

Account Package

This package provides account management functionality split across multiple modules for better organization.

Modules:

Name	Description
`exceptions`	Exception classes for account operations
`core`	Core account retrieval functions
`oauth`	OAuth-specific account functions
`config`	Account configuration management
`crud`	Create, read, update, delete operations
`sharing`	Account sharing and access management

Access ¶

Bases: DomoEnumMixin

abc for the concept of managing access levels to Domo entities.

AccountAccess ¶

Bases: Access, Enum

v2 API sharing permissions (users and groups).

AccountAccess_v1 ¶

Bases: Access, Enum

Legacy v1 API sharing permissions (users only).

AccountNoMatchError ¶

AccountNoMatchError(
    account_id: str | None = None,
    res=None,
    message: str = "Account not found -- has it been shared with the user?",
    **kwargs
)

Bases: RouteError

Raised when a specific account cannot be found or accessed.

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

def __init__(
    self,
    account_id: str | None = None,
    res=None,
    message: str = "Account not found -- has it been shared with the user?",
    **kwargs,
):
    super().__init__(message=message, entity_id=account_id, res=res, **kwargs)

AccountSharing_Error ¶

AccountSharing_Error(
    operation: str,
    account_id: str | None = None,
    res=None,
    message: str | None = None,
    **kwargs
)

Bases: RouteError

Raised when account sharing operations fail.

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

def __init__(
    self,
    operation: str,
    account_id: str | None = None,
    res=None,
    message: str | None = None,
    **kwargs,
):
    if not message:
        message = f"Account sharing {operation} failed"
    super().__init__(message=message, entity_id=account_id, res=res, **kwargs)

Account_CRUD_Error ¶

Account_CRUD_Error(
    operation: str = "CRUD",
    account_id: str | None = None,
    res=None,
    message: str | None = None,
    **kwargs
)

Bases: RouteError

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

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

def __init__(
    self,
    operation: str = "CRUD",
    account_id: str | None = None,
    res=None,
    message: str | None = None,
    **kwargs,
):
    if not message:
        message = f"Account {operation} operation failed"
    super().__init__(message=message, entity_id=account_id, res=res, **kwargs)

Account_Config_Error ¶

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

Bases: RouteError

Raised when account configuration operations fail.

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

def __init__(
    self,
    account_id: str | None = None,
    res=None,
    message: str | None = None,
    **kwargs,
):
    if not message:
        message = "Account configuration operation failed"
    super().__init__(message=message, entity_id=account_id, res=res, **kwargs)

Account_CreateParams_Error ¶

Account_CreateParams_Error(message: str, **kwargs)

Bases: RouteError

Raised when account creation parameters are invalid.

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

def __init__(self, message: str, **kwargs):
    super().__init__(message=message, **kwargs)

Account_GET_Error ¶

Account_GET_Error(
    account_id: str | None = None, res=None, **kwargs
)

Bases: RouteError

Raised when account retrieval operations fail.

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

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

SearchAccountNotFoundError ¶

SearchAccountNotFoundError(
    search_criteria: str, res=None, **kwargs
)

Bases: RouteError

Raised when account search operations return no results.

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

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

create_account `async` ¶

create_account(
    auth: DomoAuth,
    account_name: str | None = None,
    data_provider_type: str | None = None,
    config_body: dict | None = None,
    payload: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create a new account.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`account_name`	`str \| None`	Name for the new account	`None`
`data_provider_type`	`str \| None`	Type of data provider for the account	`None`
`config_body`	`dict \| None`	Properly formatted AccountConfig dictionary	`None`
`payload`	`dict \| None`	Pre-built payload (overrides individual parameters)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing created account information

Raises:

Type	Description
`Account_CreateParams_Error`	If required parameters are missing
`Account_CRUD_Error`	If account creation fails

Source code in src/crew_dcs/routes/account/crud.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def create_account(
    auth: DomoAuth,
    account_name: str | None = None,
    data_provider_type: str | None = None,
    config_body: dict | None = None,
    payload: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a new account.

    Args:
        auth: Authentication object for API requests
        account_name: Name for the new account
        data_provider_type: Type of data provider for the account
        config_body: Properly formatted AccountConfig dictionary
        payload: Pre-built payload (overrides individual parameters)
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object containing created account information

    Raises:
        Account_CreateParams_Error: If required parameters are missing
        Account_CRUD_Error: If account creation fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    if not payload and not (account_name and data_provider_type):
        raise Account_CreateParams_Error(
            "Either payload must be provided or both account_name and data_provider_type are required"
        )

    payload = payload or generate_create_account_body(
        account_name=account_name,
        data_provider_type=data_provider_type,
        config_body=config_body,
    )
    url = f"https://{auth.domo_instance}.domo.com/api/data/v1/accounts"

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

    if return_raw:
        return res

    if not res.is_success:
        raise Account_CRUD_Error(
            operation="create",
            account_id=account_name,
            res=res,
        )

    return res

create_oauth_account `async` ¶

create_oauth_account(
    auth: DomoAuth,
    account_name: str | None = None,
    data_provider_type: str | None = None,
    origin: str = "OAUTH_CONFIGURATION",
    config: dict | None = None,
    create_body: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create a new OAuth account.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`account_name`	`str \| None`	Name for the new OAuth account	`None`
`data_provider_type`	`str \| None`	Type of data provider for the OAuth account	`None`
`origin`	`str`	Origin type (default: "OAUTH_CONFIGURATION")	`'OAUTH_CONFIGURATION'`
`config`	`dict \| None`	OAuth configuration dictionary	`None`
`create_body`	`dict \| None`	Pre-built create body (overrides individual parameters)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing created OAuth account information

Raises:

Type	Description
`Account_CreateParams_Error`	If required parameters are missing
`Account_CRUD_Error`	If OAuth account creation fails

Source code in src/crew_dcs/routes/account/crud.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def create_oauth_account(
    auth: DomoAuth,
    account_name: str | None = None,
    data_provider_type: str | None = None,
    origin: str = "OAUTH_CONFIGURATION",
    config: dict | None = None,
    create_body: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a new OAuth account.

    Args:
        auth: Authentication object for API requests
        account_name: Name for the new OAuth account
        data_provider_type: Type of data provider for the OAuth account
        origin: Origin type (default: "OAUTH_CONFIGURATION")
        config: OAuth configuration dictionary
        create_body: Pre-built create body (overrides individual parameters)
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object containing created OAuth account information

    Raises:
        Account_CreateParams_Error: If required parameters are missing
        Account_CRUD_Error: If OAuth account creation fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

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

    if not create_body and not (
        account_name and data_provider_type and origin and config
    ):
        raise Account_CreateParams_Error(
            "If not passing complete create_body must pass account_name, data_provider_type, origin, and config"
        )

    create_body = create_body or generate_create_oauth_account_body(
        account_name=account_name,
        data_provider_type=data_provider_type,
        origin=origin,
        config=config,
    )

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

    if return_raw:
        return res

    if not res.is_success:
        raise Account_CRUD_Error(
            operation="create",
            account_id=create_body.get("displayName"),
            res=res,
        )

    return res

delete_account `async` ¶

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

Delete an account.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`account_id`	`str`	ID of the account to delete	required
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object confirming deletion

Raises:

Type	Description
`Account_CRUD_Error`	If account deletion fails

Source code in src/crew_dcs/routes/account/crud.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def delete_account(
    auth: DomoAuth,
    account_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Delete an account.

    Args:
        auth: Authentication object for API requests
        account_id: ID of the account to delete
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object confirming deletion

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

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

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

    if return_raw:
        return res

    if not res.is_success:
        raise Account_CRUD_Error(
            operation="delete",
            account_id=account_id,
            res=res,
        )

    return res

delete_oauth_account `async` ¶

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

Delete an OAuth account.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`account_id`	`str`	ID of the OAuth account to delete	required
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object confirming deletion

Raises:

Type	Description
`Account_CRUD_Error`	If OAuth account deletion fails

Source code in src/crew_dcs/routes/account/crud.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def delete_oauth_account(
    auth: DomoAuth,
    account_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Delete an OAuth account.

    Args:
        auth: Authentication object for API requests
        account_id: ID of the OAuth account to delete
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object confirming deletion

    Raises:
        Account_CRUD_Error: If OAuth account deletion fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/data/v1/accounts/templates/{account_id}"

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

    if return_raw:
        return res

    if not res.is_success:
        raise Account_CRUD_Error(operation="delete", account_id=account_id, res=res)

    res.response = f"deleted account {account_id}"
    return res

generate_create_account_body ¶

generate_create_account_body(
    account_name: str,
    data_provider_type: str,
    config_body: dict,
) -> dict

Generate payload for account creation.

Source code in src/crew_dcs/routes/account/crud.py

def generate_create_account_body(
    account_name: str,
    data_provider_type: str,
    config_body: dict,
) -> dict:
    """Generate payload for account creation."""
    return {
        "displayName": account_name,
        "dataProviderType": data_provider_type,
        "name": account_name,
        "configurations": config_body,
    }

generate_create_oauth_account_body ¶

generate_create_oauth_account_body(
    account_name: str,
    data_provider_type: str,
    origin: str,
    config: dict,
) -> dict

Generate payload for OAuth account creation.

Source code in src/crew_dcs/routes/account/crud.py

def generate_create_oauth_account_body(
    account_name: str, data_provider_type: str, origin: str, config: dict
) -> dict:
    """Generate payload for OAuth account creation."""
    return {
        "name": account_name,
        "displayName": account_name,
        "dataProviderType": data_provider_type,
        "origin": origin,
        "configurations": config,
    }

generate_share_account_payload ¶

generate_share_account_payload(
    access_level: AccountAccess | AccountAccess_v1 | str,
    user_id: int | None = None,
    group_id: int | None = None,
    use_v1_api: bool = False,
) -> dict

Orchestrator function to generate appropriate sharing payload.

This function determines which payload generation function to use based on: - The access_level type (v1 or v2 enum) - The use_v1_api flag - Whether a group_id is provided (forces v2)

Parameters:

Name	Type	Description	Default
`access_level`	`AccountAccess \| AccountAccess_v1 \| str`	Access level enum or string	required
`user_id`	`int \| None`	ID of the user to share with	`None`
`group_id`	`int \| None`	ID of the group to share with (forces v2 API)	`None`
`use_v1_api`	`bool`	Force use of v1 API (ignored if group_id provided)	`False`

Returns:

Type	Description
`dict`	Dictionary payload for appropriate share API

Raises:

Type	Description
`ValueError`	If group_id provided with v1 access level or use_v1_api=True

Example

V2 with user¶

payload = generate_share_account_payload( ... access_level=AccountAccess.CAN_VIEW, ... user_id=12345 ... )

V2 with group (auto-detects v2 needed)¶

payload = generate_share_account_payload( ... access_level=AccountAccess.CAN_VIEW, ... group_id=67890 ... )

V1 explicitly¶

payload = generate_share_account_payload( ... access_level=AccountAccess_v1.CAN_VIEW, ... user_id=12345 ... )

Source code in src/crew_dcs/routes/account/access.py

def generate_share_account_payload(
    access_level: AccountAccess | AccountAccess_v1 | str,
    user_id: int | None = None,
    group_id: int | None = None,
    use_v1_api: bool = False,
) -> dict:
    """Orchestrator function to generate appropriate sharing payload.

    This function determines which payload generation function to use based on:
    - The access_level type (v1 or v2 enum)
    - The use_v1_api flag
    - Whether a group_id is provided (forces v2)

    Args:
        access_level: Access level enum or string
        user_id: ID of the user to share with
        group_id: ID of the group to share with (forces v2 API)
        use_v1_api: Force use of v1 API (ignored if group_id provided)

    Returns:
        Dictionary payload for appropriate share API

    Raises:
        ValueError: If group_id provided with v1 access level or use_v1_api=True

    Example:
        >>> # V2 with user
        >>> payload = generate_share_account_payload(
        ...     access_level=AccountAccess.CAN_VIEW,
        ...     user_id=12345
        ... )

        >>> # V2 with group (auto-detects v2 needed)
        >>> payload = generate_share_account_payload(
        ...     access_level=AccountAccess.CAN_VIEW,
        ...     group_id=67890
        ... )

        >>> # V1 explicitly
        >>> payload = generate_share_account_payload(
        ...     access_level=AccountAccess_v1.CAN_VIEW,
        ...     user_id=12345
        ... )
    """
    # Determine if we need v1 or v2
    is_v1_enum = isinstance(access_level, AccountAccess_v1)

    # Force v2 if group_id provided
    if group_id:
        if is_v1_enum:
            raise ValueError("Cannot share with groups using v1 access levels")
        if use_v1_api:
            raise ValueError("Cannot share with groups using v1 API")

        return generate_share_account_v2_payload(
            access_level=access_level, group_id=group_id
        )

    # Use v1 if explicitly requested or v1 enum provided
    if is_v1_enum or use_v1_api:
        if not user_id:
            raise ValueError("user_id required for v1 API")

        # Convert v2 enum to v1 if needed
        if not is_v1_enum:
            # Map v2 access levels to v1
            v2_to_v1_mapping = {
                "CAN_VIEW": AccountAccess_v1.CAN_VIEW,
                "CAN_EDIT": AccountAccess_v1.CAN_EDIT,
                "CAN_SHARE": AccountAccess_v1.OWNER,
                "OWNER": AccountAccess_v1.OWNER,
            }

            if isinstance(access_level, str):
                access_level = AccountAccess.get(access_level)

            access_level = v2_to_v1_mapping.get(
                access_level.value, AccountAccess_v1.CAN_VIEW
            )

        return generate_share_account_v1_payload(
            user_id=user_id, access_level=access_level
        )

    # Default to v2
    return generate_share_account_v2_payload(access_level=access_level, user_id=user_id)

generate_share_account_v1_payload ¶

generate_share_account_v1_payload(
    user_id: int, access_level: AccountAccess_v1
) -> dict

Generate v1 API sharing payload for users only.

V1 API limitations: - Only supports sharing with users (no groups) - Limited permission set: READ, WRITE, OWNER

Parameters:

Name	Type	Description	Default
`user_id`	`int`	ID of the user to share with	required
`access_level`	`AccountAccess_v1`	Access level (AccountAccess_v1 enum)	required

Returns:

Type	Description
`dict`	Dictionary payload for v1 share API

Example

payload = generate_share_account_v1_payload( ... user_id=12345, ... access_level=AccountAccess_v1.CAN_VIEW ... )

Returns: {"type": "USER", "id": 12345, "permissions": ["READ"]}¶

Source code in src/crew_dcs/routes/account/access.py

def generate_share_account_v1_payload(
    user_id: int,
    access_level: AccountAccess_v1,
) -> dict:
    """Generate v1 API sharing payload for users only.

    V1 API limitations:
    - Only supports sharing with users (no groups)
    - Limited permission set: READ, WRITE, OWNER

    Args:
        user_id: ID of the user to share with
        access_level: Access level (AccountAccess_v1 enum)

    Returns:
        Dictionary payload for v1 share API

    Example:
        >>> payload = generate_share_account_v1_payload(
        ...     user_id=12345,
        ...     access_level=AccountAccess_v1.CAN_VIEW
        ... )
        >>> # Returns: {"type": "USER", "id": 12345, "permissions": ["READ"]}
    """
    return {"type": "USER", "id": int(user_id), "permissions": [access_level.value]}

generate_share_account_v2_payload ¶

generate_share_account_v2_payload(
    access_level: AccountAccess,
    user_id: int | None = None,
    group_id: int | None = None,
) -> dict

Generate v2 API sharing payload for users or groups.

V2 API features: - Supports both users and groups - Extended permissions: CAN_VIEW, CAN_EDIT, CAN_SHARE, OWNER, NO_ACCESS

Parameters:

Name	Type	Description	Default
`access_level`	`AccountAccess`	Access level (AccountAccess enum)	required
`user_id`	`int \| None`	ID of the user to share with (mutually exclusive with group_id)	`None`
`group_id`	`int \| None`	ID of the group to share with (mutually exclusive with user_id)	`None`

Returns:

Type	Description
`dict`	Dictionary payload for v2 share API

Raises:

Type	Description
`ValueError`	If neither or both user_id and group_id are provided

Example

payload = generate_share_account_v2_payload( ... access_level=AccountAccess.CAN_VIEW, ... user_id=12345 ... )

Returns: {"type": "USER", "id": "12345", "accessLevel": "CAN_VIEW"}¶

Source code in src/crew_dcs/routes/account/access.py

def generate_share_account_v2_payload(
    access_level: AccountAccess,
    user_id: int | None = None,
    group_id: int | None = None,
) -> dict:
    """Generate v2 API sharing payload for users or groups.

    V2 API features:
    - Supports both users and groups
    - Extended permissions: CAN_VIEW, CAN_EDIT, CAN_SHARE, OWNER, NO_ACCESS

    Args:
        access_level: Access level (AccountAccess enum)
        user_id: ID of the user to share with (mutually exclusive with group_id)
        group_id: ID of the group to share with (mutually exclusive with user_id)

    Returns:
        Dictionary payload for v2 share API

    Raises:
        ValueError: If neither or both user_id and group_id are provided

    Example:
        >>> payload = generate_share_account_v2_payload(
        ...     access_level=AccountAccess.CAN_VIEW,
        ...     user_id=12345
        ... )
        >>> # Returns: {"type": "USER", "id": "12345", "accessLevel": "CAN_VIEW"}
    """
    if not user_id and not group_id:
        raise ValueError("Must provide either user_id or group_id")

    if user_id and group_id:
        raise ValueError("Cannot provide both user_id and group_id")

    if user_id:
        return {"type": "USER", "id": str(user_id), "accessLevel": access_level.value}

    return {"type": "GROUP", "id": int(group_id), "accessLevel": access_level.value}

get_account_accesslist `async` ¶

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

Get access list for an account.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`account_id`	`str`	ID of the account to get access list for	required
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing account access list

Raises:

Type	Description
`AccountSharing_Error`	If access list retrieval fails

Source code in src/crew_dcs/routes/account/access.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def get_account_accesslist(
    auth: DomoAuth,
    account_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get access list for an account.

    Args:
        auth: Authentication object for API requests
        account_id: ID of the account to get access list for
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object containing account access list

    Raises:
        AccountSharing_Error: If access list retrieval fails
    """

    context = RouteContext.build_context(context=context, **context_kwargs)

    url = (
        f"https://{auth.domo_instance}.domo.com/api/data/v2/accounts/share/{account_id}"
    )

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

    if return_raw:
        return res

    if not res.is_success:
        raise AccountSharing_Error(
            operation="get access list", account_id=account_id, res=res
        )

    res.response = res.response["list"]

    return res

get_account_by_id `async` ¶

get_account_by_id(
    auth: DomoAuth,
    account_id: int | str,
    is_unmask: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve metadata about a specific account.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`account_id`	`int \| str`	The ID of the account to retrieve	required
`is_unmask`	`bool`	Whether to unmask encrypted values in response	`False`
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing account metadata

Raises:

Type	Description
`AccountNoMatchError`	If account is not found or not accessible
`Account_GET_Error`	If account retrieval fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def get_account_by_id(
    auth: DomoAuth,
    account_id: int | str,
    is_unmask: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve metadata about a specific account.

    Args:
        auth: Authentication object for API requests
        account_id: The ID of the account to retrieve
        is_unmask: Whether to unmask encrypted values in response
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object containing account metadata

    Raises:
        AccountNoMatchError: If account is not found or not accessible
        Account_GET_Error: If account retrieval fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

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

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        timeout=20,  # occasionally this API has a long response time
        params={"unmask": is_unmask},
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success and (
        res.response == "Forbidden" or res.response == "Not Found"
    ):
        raise AccountNoMatchError(account_id=str(account_id), res=res)

    if not res.is_success:
        raise Account_GET_Error(account_id=str(account_id), res=res)

    return res

get_account_config `async` ¶

get_account_config(
    auth: DomoAuth,
    account_id: int | str,
    data_provider_type: str | None = None,
    is_unmask: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve configuration for a specific account.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`account_id`	`int \| str`	The ID of the account to get config for	required
`data_provider_type`	`str \| None`	Type of data provider (auto-detected if not provided)	`None`
`is_unmask`	`bool`	Whether to unmask encrypted values in config	`True`
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing account configuration

Raises:

Type	Description
`AccountNoMatchError`	If account is not found or not accessible
`Account_Config_Error`	If account configuration retrieval fails

Source code in src/crew_dcs/routes/account/config.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def get_account_config(
    auth: DomoAuth,
    account_id: int | str,
    data_provider_type: str | None = None,
    is_unmask: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve configuration for a specific account.

    Args:
        auth: Authentication object for API requests
        account_id: The ID of the account to get config for
        data_provider_type: Type of data provider (auto-detected if not provided)
        is_unmask: Whether to unmask encrypted values in config
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object containing account configuration

    Raises:
        AccountNoMatchError: If account is not found or not accessible
        Account_Config_Error: If account configuration retrieval fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    if not data_provider_type:
        # Reuse the same RouteContext for the metadata lookup
        res = await get_account_by_id(
            auth=auth,
            account_id=account_id,
            is_unmask=is_unmask,
            context=context,
            return_raw=True,
        )
        data_provider_type = res.response["dataProviderType"]

    url = f"https://{auth.domo_instance}.domo.com/api/data/v1/providers/{data_provider_type}/account/{account_id}"

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

    if return_raw:
        return res

    if not res.is_success and (
        res.response == "Forbidden" or res.response == "Not Found"
    ):
        raise AccountNoMatchError(account_id=str(account_id), res=res)

    if not res.is_success:
        raise Account_Config_Error(account_id=str(account_id), res=res)

    res.response.update(
        {
            "_search_metadata": {
                "account_id": account_id,
                "data_provider_type": data_provider_type,
            }
        }
    )

    return res

get_accounts `async` ¶

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

Retrieve a list of all accounts the user has read access to.

Note: Users with "Manage all accounts" permission will retrieve all account objects.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing account list

Raises:

Type	Description
`Account_GET_Error`	If account retrieval fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def get_accounts(
    auth: DomoAuth,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve a list of all accounts the user has read access to.

    Note: Users with "Manage all accounts" permission will retrieve all account objects.

    Args:
        auth: Authentication object for API requests
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object containing account list

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

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

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

    if return_raw:
        return res

    if not res.is_success:
        raise Account_GET_Error(res=res)
    return res

get_available_data_providers `async` ¶

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

Retrieve available data providers from Domo.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing available data providers

Raises:

Type	Description
`Account_GET_Error`	If data provider retrieval fails

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

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def get_available_data_providers(
    auth: DomoAuth,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve available data providers from Domo.

    Args:
        auth: Authentication object
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object containing available data providers

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

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

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

    if return_raw:
        return res

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

    return res

get_oauth_account_accesslist `async` ¶

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

Get access list for an OAuth account.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`account_id`	`str`	ID of the OAuth account to get access list for	required
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing OAuth account access list

Raises:

Type	Description
`AccountSharing_Error`	If OAuth access list retrieval fails

Source code in src/crew_dcs/routes/account/access.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def get_oauth_account_accesslist(
    auth: DomoAuth,
    account_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get access list for an OAuth account.

    Args:
        auth: Authentication object for API requests
        account_id: ID of the OAuth account to get access list for
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object containing OAuth account access list

    Raises:
        AccountSharing_Error: If OAuth access list retrieval fails
    """

    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/data/v2/accounts/templates/{account_id}/share"

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

    if return_raw:
        return res

    if not res.is_success:
        raise AccountSharing_Error(
            operation="get OAuth access list", account_id=account_id, res=res
        )

    return res

get_oauth_account_by_id `async` ¶

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

Retrieve a specific OAuth account by ID.

Note: This function retrieves all OAuth accounts and filters to the selected one, as there doesn't appear to be a direct API for retrieving a single OAuth account.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`account_id`	`int \| str`	The ID of the OAuth account to retrieve	required
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing OAuth account metadata

Raises:

Type	Description
`AccountNoMatchError`	If OAuth account is not found
`Account_GET_Error`	If OAuth account retrieval fails

Source code in src/crew_dcs/routes/account/oauth.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def get_oauth_account_by_id(
    auth: DomoAuth,
    account_id: int | str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve a specific OAuth account by ID.

    Note: This function retrieves all OAuth accounts and filters to the selected one,
    as there doesn't appear to be a direct API for retrieving a single OAuth account.

    Args:
        auth: Authentication object for API requests
        account_id: The ID of the OAuth account to retrieve
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object containing OAuth account metadata

    Raises:
        AccountNoMatchError: If OAuth account is not found
        Account_GET_Error: If OAuth account retrieval fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await get_oauth_accounts(
        auth=auth,
        return_raw=return_raw,
        context=context,
    )

    if return_raw:
        return res

    # Convert account_id to int for comparison if it's a string
    target_id = int(account_id) if isinstance(account_id, str) else account_id
    res.response = next((obj for obj in res.response if obj["id"] == target_id), None)

    if not res.response:
        raise AccountNoMatchError(account_id=str(account_id), res=res)

    return res

get_oauth_account_config `async` ¶

get_oauth_account_config(
    auth: DomoAuth,
    account_id: int | str,
    data_provider_type: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve configuration for a specific OAuth account.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`account_id`	`int \| str`	The ID of the OAuth account to get config for	required
`data_provider_type`	`str`	Type of data provider for the OAuth account	required
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing OAuth account configuration

Raises:

Type	Description
`AccountNoMatchError`	If OAuth account is not found or not accessible
`Account_Config_Error`	If OAuth account configuration retrieval fails

Source code in src/crew_dcs/routes/account/config.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def get_oauth_account_config(
    auth: DomoAuth,
    account_id: int | str,
    data_provider_type: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve configuration for a specific OAuth account.

    Args:
        auth: Authentication object for API requests
        account_id: The ID of the OAuth account to get config for
        data_provider_type: Type of data provider for the OAuth account
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object containing OAuth account configuration

    Raises:
        AccountNoMatchError: If OAuth account is not found or not accessible
        Account_Config_Error: If OAuth account configuration retrieval fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/data/v1/providers/{data_provider_type}/template/{account_id}?unmask=true"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        timeout=20,  # occasionally this API has a long response time
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success and (
        res.response == "Forbidden" or res.response == "Not Found"
    ):
        raise AccountNoMatchError(account_id=str(account_id), res=res)

    if not res.is_success:
        raise Account_Config_Error(account_id=str(account_id), res=res)

    res.response.update(
        {
            "_search_metadata": {
                "account_id": account_id,
                "data_provider_type": data_provider_type,
            }
        }
    )

    return res

get_oauth_accounts `async` ¶

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

Retrieve all OAuth accounts the user has access to.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing OAuth account list

Raises:

Type	Description
`Account_GET_Error`	If OAuth account retrieval fails

Source code in src/crew_dcs/routes/account/oauth.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def get_oauth_accounts(
    auth: DomoAuth,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve all OAuth accounts the user has access to.

    Args:
        auth: Authentication object for API requests
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object containing OAuth account list

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

    url = f"https://{auth.domo_instance}.domo.com/api/data/v1/accounts/templates/user/extended"

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

    if return_raw:
        return res

    if not res.is_success:
        raise Account_GET_Error(res=res)
    return res

share_account `async` ¶

share_account(
    auth: DomoAuth,
    account_id: str,
    share_payload: dict | None = None,
    access_level: (
        Access | AccountAccess_v1 | str | None
    ) = None,
    user_id: int | None = None,
    group_id: int | None = None,
    use_v1_api: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Share account with users/groups using v2 API.

Note: This uses the v2 API which should be deployed to all Domo instances as of DP24.

Two ways to call this function: 1. Pass pre-built share_payload (legacy mode) 2. Pass access_level with user_id or group_id (generates payload automatically)

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`account_id`	`str`	ID of the account to share	required
`share_payload`	`dict \| None`	Pre-built sharing payload (optional, overrides other params)	`None`
`access_level`	`Access \| AccountAccess_v1 \| str \| None`	Access level enum or string (required if share_payload not provided)	`None`
`user_id`	`int \| None`	ID of user to share with (mutually exclusive with group_id)	`None`
`group_id`	`int \| None`	ID of group to share with (mutually exclusive with user_id)	`None`
`use_v1_api`	`bool`	Force v1 API usage (ignored if group_id provided)	`False`
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object confirming sharing operation

Raises:

Type	Description
`ValueError`	If neither share_payload nor access_level provided
`Account_AlreadyShared_Error`	If account is already shared with target
`Account_Share_Error`	If sharing operation fails

Examples:

>>> # Using payload generator (new way)
>>> await share_account(
...     auth=auth,
...     account_id="123",
...     access_level=ShareAccount_AccessLevel.CAN_VIEW,
...     user_id=456
... )

>>> # Using pre-built payload (legacy way)
>>> payload = generate_share_account_v2_payload(
...     access_level=ShareAccount_AccessLevel.CAN_VIEW,
...     group_id=789
... )
>>> await share_account(
...     auth=auth,
...     account_id="123",
...     share_payload=payload
... )

Source code in src/crew_dcs/routes/account/access.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def share_account(
    auth: DomoAuth,
    account_id: str,
    share_payload: dict | None = None,
    access_level: Access | AccountAccess_v1 | str | None = None,
    user_id: int | None = None,
    group_id: int | None = None,
    use_v1_api: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Share account with users/groups using v2 API.

    Note: This uses the v2 API which should be deployed to all Domo instances as of DP24.

    Two ways to call this function:
    1. Pass pre-built share_payload (legacy mode)
    2. Pass access_level with user_id or group_id (generates payload automatically)

    Args:
        auth: Authentication object for API requests
        account_id: ID of the account to share
        share_payload: Pre-built sharing payload (optional, overrides other params)
        access_level: Access level enum or string (required if share_payload not provided)
        user_id: ID of user to share with (mutually exclusive with group_id)
        group_id: ID of group to share with (mutually exclusive with user_id)
        use_v1_api: Force v1 API usage (ignored if group_id provided)
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object confirming sharing operation

    Raises:
        ValueError: If neither share_payload nor access_level provided
        Account_AlreadyShared_Error: If account is already shared with target
        Account_Share_Error: If sharing operation fails

    Examples:
        >>> # Using payload generator (new way)
        >>> await share_account(
        ...     auth=auth,
        ...     account_id="123",
        ...     access_level=ShareAccount_AccessLevel.CAN_VIEW,
        ...     user_id=456
        ... )

        >>> # Using pre-built payload (legacy way)
        >>> payload = generate_share_account_v2_payload(
        ...     access_level=ShareAccount_AccessLevel.CAN_VIEW,
        ...     group_id=789
        ... )
        >>> await share_account(
        ...     auth=auth,
        ...     account_id="123",
        ...     share_payload=payload
        ... )
    """

    context = RouteContext.build_context(context=context, **context_kwargs)

    # Generate payload if not provided
    if not share_payload:
        if not access_level:
            raise ValueError("Must provide either share_payload or access_level")

        share_payload = generate_share_account_payload(
            access_level=access_level,
            user_id=user_id,
            group_id=group_id,
            use_v1_api=use_v1_api,
        )

    # Route to appropriate API endpoint
    if use_v1_api or (access_level and isinstance(access_level, AccountAccess_v1)):
        # Use v1 endpoint
        url = f"https://{auth.domo_instance}.domo.com/api/data/v1/accounts/{account_id}/share"
    else:
        # Use v2 endpoint
        url = f"https://{auth.domo_instance}.domo.com/api/data/v2/accounts/share/{account_id}"

    method = "PUT"

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

    if return_raw:
        return res

    if res.status == 500 and res.response == "Internal Server Error":
        raise AccountSharing_Error(
            account_id=account_id,
            message=f"{res.response} - User may already have access to account",
            res=res,
            operation=method,
        )

    if not res.is_success:
        raise AccountSharing_Error(account_id=account_id, res=res, operation=method)

    return res

share_account_v1 `async` ¶

share_account_v1(
    auth: DomoAuth,
    account_id: str,
    share_payload: dict | None = None,
    access_level: AccountAccess_v1 | str | None = None,
    user_id: int | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Share account using legacy v1 API (users only).

Note: V1 API allows sharing with users ONLY. It does not support sharing with groups and has a more limited set of share rights (owner or read). See AccountAccess_v1 vs AccountAccess for differences.

Two ways to call this function: 1. Pass pre-built share_payload (legacy mode) 2. Pass access_level with user_id (generates payload automatically)

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`account_id`	`str`	ID of the account to share	required
`share_payload`	`dict \| None`	Pre-built sharing payload (optional, overrides other params)	`None`
`access_level`	`AccountAccess_v1 \| str \| None`	Access level enum or string (required if share_payload not provided)	`None`
`user_id`	`int \| None`	ID of user to share with (required if share_payload not provided)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object confirming sharing operation

Raises:

Type	Description
`ValueError`	If neither share_payload nor (access_level and user_id) provided
`Account_AlreadyShared_Error`	If account is already shared with user
`Account_Share_Error`	If sharing operation fails

Examples:

>>> # Using payload generator
>>> await share_account_v1(
...     auth=auth,
...     account_id="123",
...     access_level=ShareAccount_V1_AccessLevel.CAN_VIEW,
...     user_id=456
... )

Source code in src/crew_dcs/routes/account/access.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def share_account_v1(
    auth: DomoAuth,
    account_id: str,
    share_payload: dict | None = None,
    access_level: AccountAccess_v1 | str | None = None,
    user_id: int | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Share account using legacy v1 API (users only).

    Note: V1 API allows sharing with users ONLY. It does not support sharing with groups
    and has a more limited set of share rights (owner or read). See AccountAccess_v1
    vs AccountAccess for differences.

    Two ways to call this function:
    1. Pass pre-built share_payload (legacy mode)
    2. Pass access_level with user_id (generates payload automatically)

    Args:
        auth: Authentication object for API requests
        account_id: ID of the account to share
        share_payload: Pre-built sharing payload (optional, overrides other params)
        access_level: Access level enum or string (required if share_payload not provided)
        user_id: ID of user to share with (required if share_payload not provided)
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object confirming sharing operation

    Raises:
        ValueError: If neither share_payload nor (access_level and user_id) provided
        Account_AlreadyShared_Error: If account is already shared with user
        Account_Share_Error: If sharing operation fails

    Examples:
        >>> # Using payload generator
        >>> await share_account_v1(
        ...     auth=auth,
        ...     account_id="123",
        ...     access_level=ShareAccount_V1_AccessLevel.CAN_VIEW,
        ...     user_id=456
        ... )
    """

    context = RouteContext.build_context(context=context, **context_kwargs)

    # Generate payload if not provided
    if not share_payload:
        if not access_level or not user_id:
            raise ValueError(
                "Must provide either share_payload or (access_level and user_id)"
            )

        share_payload = generate_share_account_v1_payload(
            user_id=user_id,
            access_level=access_level,
        )

    url = (
        f"https://{auth.domo_instance}.domo.com/api/data/v1/accounts/{account_id}/share"
    )

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

    if return_raw:
        return res

    if res.status == 500 and res.response == "Internal Server Error":
        raise AccountSharing_Error(
            account_id=account_id,
            message=f"{res.response} - User may already have access to account",
            res=res,
            operation="PUT",
        )

    if not res.is_success:
        raise AccountSharing_Error(
            account_id=account_id,
            res=res,
            operation="PUT",
        )

    return res

share_oauth_account `async` ¶

share_oauth_account(
    auth: DomoAuth,
    account_id: str,
    share_payload: dict | None = None,
    access_level: Access | str | None = None,
    user_id: int | None = None,
    group_id: int | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Share OAuth account with users/groups.

Two ways to call this function: 1. Pass pre-built share_payload (legacy mode) 2. Pass access_level with user_id or group_id (generates payload automatically)

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`account_id`	`str`	ID of the OAuth account to share	required
`share_payload`	`dict \| None`	Pre-built sharing payload (optional, overrides other params)	`None`
`access_level`	`Access \| str \| None`	Access level enum or string (required if share_payload not provided)	`None`
`user_id`	`int \| None`	ID of user to share with (mutually exclusive with group_id)	`None`
`group_id`	`int \| None`	ID of group to share with (mutually exclusive with user_id)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object confirming sharing operation

Raises:

Type	Description
`ValueError`	If neither share_payload nor access_level provided
`Account_Share_Error`	If OAuth sharing operation fails

Examples:

>>> # Using payload generator
>>> await share_oauth_account(
...     auth=auth,
...     account_id="123",
...     access_level=ShareAccount_AccessLevel.CAN_VIEW,
...     group_id=789
... )

Source code in src/crew_dcs/routes/account/access.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def share_oauth_account(
    auth: DomoAuth,
    account_id: str,
    share_payload: dict | None = None,
    access_level: Access | str | None = None,
    user_id: int | None = None,
    group_id: int | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Share OAuth account with users/groups.

    Two ways to call this function:
    1. Pass pre-built share_payload (legacy mode)
    2. Pass access_level with user_id or group_id (generates payload automatically)

    Args:
        auth: Authentication object for API requests
        account_id: ID of the OAuth account to share
        share_payload: Pre-built sharing payload (optional, overrides other params)
        access_level: Access level enum or string (required if share_payload not provided)
        user_id: ID of user to share with (mutually exclusive with group_id)
        group_id: ID of group to share with (mutually exclusive with user_id)
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object confirming sharing operation

    Raises:
        ValueError: If neither share_payload nor access_level provided
        Account_Share_Error: If OAuth sharing operation fails

    Examples:
        >>> # Using payload generator
        >>> await share_oauth_account(
        ...     auth=auth,
        ...     account_id="123",
        ...     access_level=ShareAccount_AccessLevel.CAN_VIEW,
        ...     group_id=789
        ... )
    """

    context = RouteContext.build_context(context=context, **context_kwargs)

    # Generate payload if not provided
    if not share_payload:
        if not access_level:
            raise ValueError("Must provide either share_payload or access_level")

        share_payload = generate_share_account_v2_payload(
            access_level=access_level,
            user_id=user_id,
            group_id=group_id,
        )

    url = f"https://{auth.domo_instance}.domo.com/api/data/v2/accounts/templates/share/{account_id}"

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

    if return_raw:
        return res

    if not res.is_success:
        raise AccountSharing_Error(
            account_id=account_id,
            res=res,
            operation="share OAuth account",
        )

    return res

update_account_config `async` ¶

update_account_config(
    auth: DomoAuth,
    account_id: int | str,
    config_body: dict,
    data_provider_type: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update configuration for an account.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`account_id`	`int \| str`	ID of the account to update config for	required
`config_body`	`dict`	New configuration data	required
`data_provider_type`	`str \| None`	Type of data provider (auto-detected if not provided)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object confirming config update

Raises:

Type	Description
`Account_Config_Error`	If account configuration update fails

Source code in src/crew_dcs/routes/account/config.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def update_account_config(
    auth: DomoAuth,
    account_id: int | str,
    config_body: dict,
    data_provider_type: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update configuration for an account.

    Args:
        auth: Authentication object for API requests
        account_id: ID of the account to update config for
        config_body: New configuration data
        data_provider_type: Type of data provider (auto-detected if not provided)
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object confirming config update

    Raises:
        Account_Config_Error: If account configuration update fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    # get the data_provider_type, which is necessary for updating the config setting
    if not data_provider_type:
        res = await get_account_by_id(
            auth=auth,
            account_id=account_id,
            is_unmask=False,
            context=context,
            return_raw=True,
        )
        data_provider_type = res.response.get("dataProviderType")

    url = f"https://{auth.domo_instance}.domo.com/api/data/v1/providers/{data_provider_type}/account/{account_id}"

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

    if return_raw:
        return res

    if res.status == 400 and res.response == "Bad Request":
        raise Account_Config_Error(
            account_id=str(account_id),
            res=res,
            message=f"Error updating config | use debug_api = True - {res.response}",
        )

    if not res.is_success:
        raise Account_Config_Error(account_id=str(account_id), res=res)

    return res

update_account_name `async` ¶

update_account_name(
    auth: DomoAuth,
    account_id: int | str,
    account_name: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update the name of an account.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`account_id`	`int \| str`	ID of the account to rename	required
`account_name`	`str`	New name for the account	required
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object confirming name update

Raises:

Type	Description
`Account_CRUD_Error`	If account name update fails

Source code in src/crew_dcs/routes/account/crud.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def update_account_name(
    auth: DomoAuth,
    account_id: int | str,
    account_name: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update the name of an account.

    Args:
        auth: Authentication object for API requests
        account_id: ID of the account to rename
        account_name: New name for the account
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object confirming name update

    Raises:
        Account_CRUD_Error: If account name update fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = (
        f"https://{auth.domo_instance}.domo.com/api/data/v1/accounts/{account_id}/name"
    )

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="PUT",
        body=account_name,
        content_type="text/plain",
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        raise Account_CRUD_Error(
            operation="update name", account_id=str(account_id), res=res
        )

    return res

update_oauth_account_config `async` ¶

update_oauth_account_config(
    auth: DomoAuth,
    account_id: int | str,
    config_body: dict,
    data_provider_type: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update configuration for an OAuth account.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`account_id`	`int \| str`	ID of the OAuth account to update config for	required
`config_body`	`dict`	New configuration data	required
`data_provider_type`	`str \| None`	Type of data provider (auto-detected if not provided)	`None`
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object confirming config update

Raises:

Type	Description
`Account_Config_Error`	If OAuth account configuration update fails

Source code in src/crew_dcs/routes/account/config.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def update_oauth_account_config(
    auth: DomoAuth,
    account_id: int | str,
    config_body: dict,
    data_provider_type: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update configuration for an OAuth account.

    Args:
        auth: Authentication object for API requests
        account_id: ID of the OAuth account to update config for
        config_body: New configuration data
        data_provider_type: Type of data provider (auto-detected if not provided)
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object confirming config update

    Raises:
        Account_Config_Error: If OAuth account configuration update fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    # get the data_provider_type, which is necessary for updating the config setting
    if not data_provider_type:
        res = await get_oauth_account_by_id(
            auth=auth,
            account_id=account_id,
            context=context,
            return_raw=True,
        )
        data_provider_type = res.response.get("dataProviderType")

    url = f"https://{auth.domo_instance}.domo.com/api/data/v1/providers/{data_provider_type}/template/{account_id}"

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

    if return_raw:
        return res

    if res.status == 400 and res.response == "Bad Request":
        raise Account_Config_Error(
            account_id=str(account_id),
            res=res,
            message=f"Error updating OAuth config | use debug_api = True - {res.response}",
        )

    if not res.is_success:
        raise Account_Config_Error(account_id=str(account_id), res=res)

    return res

update_oauth_account_name `async` ¶

update_oauth_account_name(
    auth: DomoAuth,
    account_id: int | str,
    account_name: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update the name of an OAuth account.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for API requests	required
`account_id`	`int \| str`	ID of the OAuth account to rename	required
`account_name`	`str`	New name for the OAuth account	required
`return_raw`	`bool`	Return raw response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object confirming name update

Raises:

Type	Description
`Account_CRUD_Error`	If OAuth account name update fails

Source code in src/crew_dcs/routes/account/crud.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(result_processor=ResponseGetDataProcessor()),
)
async def update_oauth_account_name(
    auth: DomoAuth,
    account_id: int | str,
    account_name: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update the name of an OAuth account.

    Args:
        auth: Authentication object for API requests
        account_id: ID of the OAuth account to rename
        account_name: New name for the OAuth account
        return_raw: Return raw response without processing
        context: RouteContext for request configuration

    Returns:
        ResponseGetData object confirming name update

    Raises:
        Account_CRUD_Error: If OAuth account name update fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/data/v1/accounts/templates/{account_id}/name"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="PUT",
        body=account_name,
        content_type="text/plain",
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        raise Account_CRUD_Error(
            operation="update name", account_id=str(account_id), res=res
        )

    return res

account

account ¶

Access ¶

AccountAccess ¶

AccountAccess_v1 ¶

AccountNoMatchError ¶

AccountSharing_Error ¶

Account_CRUD_Error ¶

Account_Config_Error ¶

Account_CreateParams_Error ¶

Account_GET_Error ¶

SearchAccountNotFoundError ¶

create_account async ¶

create_oauth_account async ¶

delete_account async ¶

delete_oauth_account async ¶

generate_create_account_body ¶

generate_create_oauth_account_body ¶

generate_share_account_payload ¶

V2 with user¶

V2 with group (auto-detects v2 needed)¶

V1 explicitly¶

generate_share_account_v1_payload ¶

Returns: {"type": "USER", "id": 12345, "permissions": ["READ"]}¶

generate_share_account_v2_payload ¶

Returns: {"type": "USER", "id": "12345", "accessLevel": "CAN_VIEW"}¶

get_account_accesslist async ¶

get_account_by_id async ¶

get_account_config async ¶

get_accounts async ¶

get_available_data_providers async ¶

get_oauth_account_accesslist async ¶

get_oauth_account_by_id async ¶

get_oauth_account_config async ¶

get_oauth_accounts async ¶

share_account async ¶

share_account_v1 async ¶

share_oauth_account async ¶

update_account_config async ¶

update_account_name async ¶

update_oauth_account_config async ¶

update_oauth_account_name async ¶

Modules¶

create_account `async` ¶

create_oauth_account `async` ¶

delete_account `async` ¶

delete_oauth_account `async` ¶

get_account_accesslist `async` ¶

get_account_by_id `async` ¶

get_account_config `async` ¶

get_accounts `async` ¶

get_available_data_providers `async` ¶

get_oauth_account_accesslist `async` ¶

get_oauth_account_by_id `async` ¶

get_oauth_account_config `async` ¶

get_oauth_accounts `async` ¶

share_account `async` ¶

share_account_v1 `async` ¶

share_oauth_account `async` ¶

update_account_config `async` ¶

update_account_name `async` ¶

update_oauth_account_config `async` ¶

update_oauth_account_name `async` ¶