account_oauth

account_oauth

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)

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

DomoAccount_OAuth dataclass

DomoAccount_OAuth(
    auth: DomoAuth,
    id: int,
    raw: dict,
    name: str = None,
    display_name: str = None,
    data_provider_type: str = None,
    created_dt: datetime = None,
    modified_dt: datetime = None,
    owners: list[Any] = None,
    is_admin_summary: bool = True,
    dataset_count: int = None,
    Config: DomoAccount_Config = None,
)

Bases: DomoAccount_Default

create async classmethod

create(
    auth: DomoAuth,
    account_name: str,
    oauth_config: OAuthConfig,
    origin: str = "OAUTH_CONFIGURATION",
    debug_api: bool = False,
    session: AsyncClient = None,
    debug_num_stacks_to_drop=2,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Create a new OAuth account.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for API requests	required
account_name	str	Display name for the OAuth account	required
oauth_config	OAuthConfig	OAuth configuration object (OAuthConfig enum member)	required
origin	str	Origin type for the OAuth account (default: "OAUTH_CONFIGURATION")	'OAUTH_CONFIGURATION'
debug_api	bool	Enable API debugging	False
session	AsyncClient	HTTP client session (optional)	None
debug_num_stacks_to_drop		Stack frames to drop for debugging	2
context	RouteContext | None	Optional RouteContext for API call configuration	None
**context_kwargs		Additional context parameters	{}

Returns:

Type	Description
	DomoAccount_OAuth instance with configuration loaded

Raises:

Type	Description
Account_CRUD_Error	If account creation fails

Source code in src/crew_dcs/classes/DomoAccount/account_oauth.py

@classmethod
async def create(
    cls,
    auth: DomoAuth,
    account_name: str,
    oauth_config: OAuthConfig,
    origin: str = "OAUTH_CONFIGURATION",
    debug_api: bool = False,
    session: httpx.AsyncClient = None,
    debug_num_stacks_to_drop=2,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Create a new OAuth account.

    Args:
        auth: Authentication object for API requests
        account_name: Display name for the OAuth account
        oauth_config: OAuth configuration object (OAuthConfig enum member)
        origin: Origin type for the OAuth account (default: "OAUTH_CONFIGURATION")
        debug_api: Enable API debugging
        session: HTTP client session (optional)
        debug_num_stacks_to_drop: Stack frames to drop for debugging
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        DomoAccount_OAuth instance with configuration loaded

    Raises:
        Account_CRUD_Error: If account creation fails
    """

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        **context_kwargs,
    )

    res = await account_routes.create_oauth_account(
        auth=auth,
        account_name=account_name,
        data_provider_type=oauth_config.data_provider_type,
        origin=origin,
        config=oauth_config.to_dict(),
        context=context,
    )

    return await cls.get_by_id(
        auth=auth,
        account_id=res.response["id"],
        context=context,
    )

delete async

delete(
    debug_api: bool = False,
    debug_num_stacks_to_drop=2,
    session: AsyncClient = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Delete this OAuth account.

Parameters:

Name	Type	Description	Default
debug_api	bool	Enable API debugging	False
debug_num_stacks_to_drop		Stack frames to drop for debugging	2
session	AsyncClient	HTTP client session (optional)	None
context	RouteContext | None	Optional RouteContext for API call configuration	None
**context_kwargs		Additional context parameters	{}

Returns:

Type	Description
	ResponseGetData object confirming deletion

Raises:

Type	Description
Account_CRUD_Error	If account deletion fails

Source code in src/crew_dcs/classes/DomoAccount/account_oauth.py

async def delete(
    self,
    debug_api: bool = False,
    debug_num_stacks_to_drop=2,
    session: httpx.AsyncClient = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Delete this OAuth account.

    Args:
        debug_api: Enable API debugging
        debug_num_stacks_to_drop: Stack frames to drop for debugging
        session: HTTP client session (optional)
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        ResponseGetData object confirming deletion

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

    return await account_routes.delete_oauth_account(
        auth=self.auth,
        account_id=self.id,
        context=context,
    )

get_by_id async classmethod

get_by_id(
    auth: DomoAuth,
    account_id: int,
    is_suppress_no_config: bool = True,
    session: AsyncClient = None,
    return_raw: bool = False,
    debug_api: bool = False,
    debug_num_stacks_to_drop=2,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Retrieve OAuth account metadata and attempt to retrieve configuration.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for API requests	required
account_id	int	ID of the OAuth account to retrieve	required
is_suppress_no_config	bool	Suppress errors when config is not defined	True
session	AsyncClient	HTTP client session (optional)	None
return_raw	bool	Return raw response without processing	False
debug_api	bool	Enable API debugging	False
debug_num_stacks_to_drop		Stack frames to drop for debugging	2
context	RouteContext | None	Optional RouteContext for API call configuration	None
**context_kwargs		Additional context parameters	{}

Returns:

Name	Type	Description
DomoAccount_OAuth		OAuth account instance with configuration

Raises:

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

Source code in src/crew_dcs/classes/DomoAccount/account_oauth.py

@classmethod
async def get_by_id(
    cls,
    auth: DomoAuth,
    account_id: int,
    is_suppress_no_config: bool = True,
    session: httpx.AsyncClient = None,
    return_raw: bool = False,
    debug_api: bool = False,
    debug_num_stacks_to_drop=2,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Retrieve OAuth account metadata and attempt to retrieve configuration.

    Args:
        auth: Authentication object for API requests
        account_id: ID of the OAuth account to retrieve
        is_suppress_no_config: Suppress errors when config is not defined
        session: HTTP client session (optional)
        return_raw: Return raw response without processing
        debug_api: Enable API debugging
        debug_num_stacks_to_drop: Stack frames to drop for debugging
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        DomoAccount_OAuth: OAuth account instance with configuration

    Raises:
        AccountNoMatchError: If OAuth account is not found
        Account_GET_Error: If OAuth account retrieval fails
        Account_Config_Error: If configuration retrieval fails
    """

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        **context_kwargs,
    )

    res = await account_routes.get_oauth_account_by_id(
        auth=auth,
        account_id=account_id,
        context=context,
    )

    if return_raw:
        return res

    acc = cls.from_dict(
        obj=res.response,
        auth=auth,
        is_admin_summary=False,
        is_use_default_account_class=False,
        new_cls=cls,
    )

    await acc._get_config(
        context=context,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop + 1,
        is_suppress_no_config=is_suppress_no_config,
    )

    return acc

share async

share(
    user_id: int | None = None,
    group_id: int | None = None,
    access_level: AccountAccess | str | None = None,
    session: AsyncClient | None = None,
    debug_api: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Share this OAuth account with a user or group.

Parameters:

Name	Type	Description	Default
user_id	int | None	User ID to share with (mutually exclusive with group_id)	None
group_id	int | None	Group ID to share with (mutually exclusive with user_id)	None
access_level	AccountAccess | str | None	Access level (ShareAccount_AccessLevel enum)	None
session	AsyncClient | None	HTTP client session (optional)	None
debug_api	bool	Enable API debugging	False
return_raw	bool	Return raw response without processing	False
context	RouteContext | None	Optional RouteContext for API call configuration	None
**context_kwargs		Additional context parameters	{}

Returns:

Type	Description
	ResponseGetData if return_raw=True, else the updated account

Raises:

Type	Description
ValueError	If neither user_id nor group_id is provided
Account_Share_Error	If sharing operation fails

Example

from crew_dcs.routes.account import AccountAccess account = await DomoAccount_OAuth.get_by_id(auth=auth, account_id=123) await account.share( ... user_id=456, ... access_level=AccountAccess.CAN_EDIT ... )

Source code in src/crew_dcs/classes/DomoAccount/account_oauth.py

async def share(
    self,
    user_id: int | None = None,
    group_id: int | None = None,
    access_level: AccountAccess | str | None = None,
    session: httpx.AsyncClient | None = None,
    debug_api: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Share this OAuth account with a user or group.

    Args:
        user_id: User ID to share with (mutually exclusive with group_id)
        group_id: Group ID to share with (mutually exclusive with user_id)
        access_level: Access level (ShareAccount_AccessLevel enum)
        session: HTTP client session (optional)
        debug_api: Enable API debugging
        return_raw: Return raw response without processing
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        ResponseGetData if return_raw=True, else the updated account

    Raises:
        ValueError: If neither user_id nor group_id is provided
        Account_Share_Error: If sharing operation fails

    Example:
        >>> from crew_dcs.routes.account import AccountAccess
        >>> account = await DomoAccount_OAuth.get_by_id(auth=auth, account_id=123)
        >>> await account.share(
        ...     user_id=456,
        ...     access_level=AccountAccess.CAN_EDIT
        ... )
    """
    if not user_id and not group_id:
        raise ValueError("Must provide either user_id or group_id")

    # Same defect as DomoAccount.share: this imported
    # `ShareAccount_AccessLevel` (no such name) and called
    # `.generate_payload()` (defined nowhere). OAuth sharing is v2-only, so
    # it uses the v2 builder directly.
    if not access_level:
        access_level = AccountAccess.CAN_VIEW

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

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

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

    res = await account_routes.share_oauth_account(
        auth=self.auth,
        account_id=self.id,
        share_payload=share_payload,
        return_raw=return_raw,
        context=context,
    )

    if return_raw:
        return res

    return self

update_config async

update_config(
    oauth_config: OAuthConfig = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop=2,
    session: AsyncClient = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Update the configuration of this OAuth account.

Parameters:

Name	Type	Description	Default
oauth_config	OAuthConfig	New OAuth configuration (defaults to current config if None)	None
debug_api	bool	Enable API debugging	False
debug_num_stacks_to_drop		Stack frames to drop for debugging	2
session	AsyncClient	HTTP client session (optional)	None
context	RouteContext | None	Optional RouteContext for API call configuration	None
**context_kwargs		Additional context parameters	{}

Returns:

Type	Description
	Self (DomoAccount_OAuth) with updated configuration

Raises:

Type	Description
Account_Config_Error	If configuration update fails

Source code in src/crew_dcs/classes/DomoAccount/account_oauth.py

async def update_config(
    self,
    oauth_config: OAuthConfig = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop=2,
    session: httpx.AsyncClient = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Update the configuration of this OAuth account.

    Args:
        oauth_config: New OAuth configuration (defaults to current config if None)
        debug_api: Enable API debugging
        debug_num_stacks_to_drop: Stack frames to drop for debugging
        session: HTTP client session (optional)
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        Self (DomoAccount_OAuth) with updated configuration

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

    await account_routes.update_oauth_account_config(
        auth=self.auth,
        account_id=self.id,
        config_body=oauth_config.to_dict() or self.Config.to_dict(),
        data_provider_type=self.data_provider_type,
        context=context,
    )

    self.Config = oauth_config

    return self

update_name async

update_name(
    account_name: str,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: AsyncClient = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Update the name of this OAuth account.

Parameters:

Name	Type	Description	Default
account_name	str	New display name for the account	required
debug_api	bool	Enable API debugging	False
debug_num_stacks_to_drop	int	Stack frames to drop for debugging	2
session	AsyncClient	HTTP client session (optional)	None
context	RouteContext | None	Optional RouteContext for API call configuration	None
**context_kwargs		Additional context parameters	{}

Returns:

Type	Description
	Self (DomoAccount_OAuth) with updated name

Raises:

Type	Description
Account_CRUD_Error	If name update fails

Source code in src/crew_dcs/classes/DomoAccount/account_oauth.py

async def update_name(
    self,
    account_name: str,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: httpx.AsyncClient = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Update the name of this OAuth account.

    Args:
        account_name: New display name for the account
        debug_api: Enable API debugging
        debug_num_stacks_to_drop: Stack frames to drop for debugging
        session: HTTP client session (optional)
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        Self (DomoAccount_OAuth) with updated name

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

    await account_routes.update_oauth_account_name(
        auth=self.auth,
        account_id=self.id,
        account_name=account_name,
        context=context,
    )

    self.name = account_name
    return self