DomoAccount

DomoAccount ¶

Account module providing Domo account management functionality.

This module contains classes for managing Domo accounts, credentials, configurations, and OAuth settings.

Classes:

Name	Description
`DomoAccount`	Main account class
`DomoAccount_Default`	Base account functionality
`DomoAccount_Credential`	Account credential management
`DomoAccount_OAuth`	OAuth account configurations
`AccountConfig`	Account configuration classes
`DomoAccounts`	Collection class for managing multiple accounts

Example

Basic account usage:

>>> from crew_dcs.classes.Account import DomoAccount
>>> account = DomoAccount.from_dict(account_data)

Working with account collections:

>>> from crew_dcs.classes.Account import DomoAccounts
>>> accounts = await DomoAccounts(auth=auth).get()

DomoAccount `dataclass` ¶

DomoAccount(
    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

from_dict `classmethod` ¶

from_dict(
    auth: DomoAuth,
    obj: dict[str, Any],
    is_admin_summary: bool = True,
    is_use_default_class: bool = False,
    new_cls: Any = None,
    **kwargs
)

converts data_v1_accounts API response into an accounts class object

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

@classmethod
def from_dict(
    cls,
    auth: DomoAuth,
    obj: dict[str, Any],
    is_admin_summary: bool = True,
    is_use_default_class: bool = False,
    new_cls: Any = None,  # Keep for compatibility with parent signature
    **kwargs,
):
    """converts data_v1_accounts API response into an accounts class object"""

    # If new_cls is explicitly provided, use it; otherwise determine from is_use_default_class
    if new_cls is None:
        if is_use_default_class:
            new_cls = cls
        elif obj.get("credentialsType") == "oauth":
            new_cls = DomoAccount_OAuth
        else:
            new_cls = DomoAccountCredential

    return super().from_dict(
        auth=auth,
        obj=obj,
        is_admin_summary=is_admin_summary,
        new_cls=new_cls,
        **kwargs,
    )

DomoAccountCredential `dataclass` ¶

DomoAccountCredential(
    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,
    target_auth: DomoAuth = None,
    target_user: DomoUser = None,
    target_access_token: DomoAccessToken = None,
    target_instance: str = None,
    is_valid_full_auth: bool = None,
    is_valid_token_auth: bool = None,
    _token_auth: DomoAuth = None,
    _full_auth: DomoAuth = None,
)

Bases: DomoAccount_Default

Account credential management class for Domo accounts.

This class extends DomoAccount_Default to provide credential management capabilities including authentication testing, password management, and access token operations.

Attributes:

Name	Type	Description
`target_instance`	`str`	Target Domo instance for credential operations
`is_valid_full_auth`	`bool`	Whether full authentication is valid
`is_valid_token_auth`	`bool`	Whether token authentication is valid
`target_auth`	`DomoAuth`	Active authentication object for target instance
`target_user`	`DomoUser`	DomoUser object for the target user
`target_access_token`	`DomoAccessToken`	Access token for the account

from_dict `classmethod` ¶

from_dict(
    obj: dict[str, Any],
    is_admin_summary: bool = True,
    auth: DomoAuth | None = None,
    is_use_default_account_class: bool = False,
    **kwargs
)

Create Account_Credential from dictionary representation.

Parameters:

Name	Type	Description	Default
`obj`	`dict[str, Any]`	Dictionary containing account data	required
`is_admin_summary`	`bool`	Whether this is an admin summary view	`True`
`auth`	`DomoAuth \| None`	Authentication object	`None`
`**kwargs`		Additional keyword arguments including target_instance	`{}`

Returns:

Type	Description
	DomoAccount_Credential instance

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

@classmethod
def from_dict(
    cls,
    obj: dict[str, Any],
    is_admin_summary: bool = True,
    auth: DomoAuth | None = None,
    is_use_default_account_class: bool = False,
    **kwargs,
):
    """Create Account_Credential from dictionary representation.

    Args:
        obj: Dictionary containing account data
        is_admin_summary: Whether this is an admin summary view
        auth: Authentication object
        **kwargs: Additional keyword arguments including target_instance

    Returns:
        DomoAccount_Credential instance
    """
    # Note: is_use_default_account_class is consumed here and not passed to parent
    # to avoid TypeError when passed to dataclass __init__
    return super().from_dict(
        obj=obj,
        is_admin_summary=is_admin_summary,
        auth=auth,
        new_cls=cls,
        target_instance=kwargs.get("target_instance"),
    )

get_target_access_token `async` ¶

get_target_access_token(
    token_name: str | None = None,
    user_email: str | None = None,
    target_auth: DomoAuth | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoAccessToken

Retrieve an access token for the target user.

Parameters:

Name	Type	Description	Default
`token_name`	`str \| None`	Name of the access token (defaults to account name)	`None`
`user_email`	`str \| None`	Email address of user (defaults to configured username)	`None`
`target_auth`	`DomoAuth \| None`	Authentication object (defaults to self.target_auth)	`None`
`debug_api`	`bool`	Enable API debugging	`False`
`session`	`AsyncClient \| None`	HTTP client session (optional)	`None`
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`
`**context_kwargs`		Additional context parameters	`{}`

Returns:

Type	Description
`DomoAccessToken`	DomoAccessToken object if found, None otherwise

Raises:

Type	Description
`DACValidAuthError`	If target authentication is not available
`DAC_NoAccessTokenName`	If token name is not provided or available

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

async def get_target_access_token(
    self,
    token_name: str | None = None,
    user_email: str | None = None,
    target_auth: DomoAuth | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoAccessToken:
    """Retrieve an access token for the target user.

    Args:
        token_name: Name of the access token (defaults to account name)
        user_email: Email address of user (defaults to configured username)
        target_auth: Authentication object (defaults to self.target_auth)
        debug_api: Enable API debugging
        session: HTTP client session (optional)
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        DomoAccessToken object if found, None otherwise

    Raises:
        DACValidAuthError: If target authentication is not available
        DAC_NoAccessTokenName: If token name is not provided or available
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    target_auth = target_auth or self.target_auth

    if not target_auth:
        raise DACValidAuthError(
            self,
            message="no target_auth, pass a valid backup_auth",
        )

    if not self.target_user:
        await self.get_target_user(
            context=context,
            user_email=user_email,
            target_auth=target_auth,
        )

    token_name = token_name or self.name

    if not token_name:
        raise DACNoAccessTokenNameError(self)

    domo_access_tokens = await self.target_user.get_access_tokens(
        context=context,
    )

    self.target_access_token = next(
        (
            dat
            for dat in domo_access_tokens
            if dat and (dat.name and dat.name.lower() == token_name.lower())
        ),
        None,
    )

    if not self.target_access_token:
        raise DACNoAccessTokenNameError(self)

    return self.target_access_token

get_target_user `async` ¶

get_target_user(
    user_email: str | None = None,
    target_auth: DomoAuth | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoUser

Retrieve the target user for this account.

Parameters:

Name	Type	Description	Default
`user_email`	`str \| None`	Email address of user (defaults to configured username)	`None`
`target_auth`	`DomoAuth \| None`	Authentication object (defaults to self.target_auth)	`None`
`debug_api`	`bool`	Enable API debugging	`False`
`session`	`AsyncClient \| None`	HTTP client session (optional)	`None`
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`

Returns:

Type	Description
`DomoUser`	DomoUser object for the target user

Raises:

Type	Description
`DAC_NoUserName`	If user email is not provided or configured
`DACValidAuthError`	If target authentication is not available
`DAC_NoTargetUser`	If user cannot be found

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

async def get_target_user(
    self,
    user_email: str | None = None,
    target_auth: DomoAuth | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoUser:
    """Retrieve the target user for this account.

    Args:
        user_email: Email address of user (defaults to configured username)
        target_auth: Authentication object (defaults to self.target_auth)
        debug_api: Enable API debugging
        session: HTTP client session (optional)
        context: Optional RouteContext for API call configuration

    Returns:
        DomoUser object for the target user

    Raises:
        DAC_NoUserName: If user email is not provided or configured
        DACValidAuthError: If target authentication is not available
        DAC_NoTargetUser: If user cannot be found
    """
    user_email = user_email or self.Config.username

    if not user_email:
        raise DACNoUserNameError(self)

    target_auth = target_auth or self.target_auth

    if not target_auth:
        raise DACValidAuthError(
            self,
            message="no target_auth, pass a valid backup_auth",
        )

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

    self.target_user = await DomoUsers(auth=target_auth).search_by_email(
        email=[user_email],
        context=context,
    )

    if not self.target_user:
        raise DACNoTargetUserError(self)

    return self.target_user

regenerate_target_access_token `async` ¶

regenerate_target_access_token(
    token_name: str | None = None,
    duration_in_days: int = 90,
    user_email: str | None = None,
    is_update_account: bool = True,
    target_auth: DomoAuth | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoAccountCredential

Regenerate or create an access token for the target user.

If a token with the given name exists, it will be regenerated. Otherwise, a new token will be created.

Parameters:

Name	Type	Description	Default
`token_name`	`str \| None`	Name of the access token (defaults to account name)	`None`
`duration_in_days`	`int`	Token validity duration in days (default: 90)	`90`
`user_email`	`str \| None`	Email address of user (defaults to configured username)	`None`
`is_update_account`	`bool`	Whether to update the account config with new token	`True`
`target_auth`	`DomoAuth \| None`	Authentication object (defaults to self.target_auth)	`None`
`debug_api`	`bool`	Enable API debugging	`False`
`session`	`AsyncClient \| None`	HTTP client session (optional)	`None`
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`
`**context_kwargs`		Additional context parameters	`{}`

Returns:

Type	Description
`DomoAccountCredential`	Self for method chaining

Raises:

Type	Description
`DACValidAuthError`	If target authentication is not available
`DAC_NoTargetUser`	If target user cannot be retrieved

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

async def regenerate_target_access_token(
    self,
    token_name: str | None = None,
    duration_in_days: int = 90,
    user_email: str | None = None,
    is_update_account: bool = True,
    target_auth: DomoAuth | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> "DomoAccountCredential":
    """Regenerate or create an access token for the target user.

    If a token with the given name exists, it will be regenerated. Otherwise,
    a new token will be created.

    Args:
        token_name: Name of the access token (defaults to account name)
        duration_in_days: Token validity duration in days (default: 90)
        user_email: Email address of user (defaults to configured username)
        is_update_account: Whether to update the account config with new token
        target_auth: Authentication object (defaults to self.target_auth)
        debug_api: Enable API debugging
        session: HTTP client session (optional)
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        Self for method chaining

    Raises:
        DACValidAuthError: If target authentication is not available
        DAC_NoTargetUser: If target user cannot be retrieved
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    target_auth = target_auth or self.target_auth

    if not target_auth:
        raise DACValidAuthError(
            self,
            message="no target_auth, pass a valid backup_auth",
        )

    domo_access_token = await self.get_target_access_token(
        token_name=token_name,
        user_email=user_email,
        target_auth=target_auth,
        context=context,
    )  # handles retrieving target user

    if not self.target_user:
        raise DACNoTargetUserError(self)

    if domo_access_token:
        await domo_access_token.regenerate(
            duration_in_days=duration_in_days, context=context
        )

    else:
        domo_access_token = await DomoAccessToken.generate(
            duration_in_days=duration_in_days,
            token_name=token_name,
            auth=target_auth,
            owner=self.target_user,
            context=context,
        )

        self.target_access_token = domo_access_token

    self.set_access_token(domo_access_token.token)

    if is_update_account:
        await self.update_config(context=context)

    return self

set_access_token ¶

set_access_token(access_token: str) -> bool

Set the access token in the account configuration.

Parameters:

Name	Type	Description	Default
`access_token`	`str`	New access token to set	required

Returns:

Type	Description
`bool`	True if successful

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

def set_access_token(self, access_token: str) -> bool:
    """Set the access token in the account configuration.

    Args:
        access_token: New access token to set

    Returns:
        True if successful
    """
    self.Config.domo_access_token = access_token
    return True

set_password ¶

set_password(password: str) -> bool

Set the password in the account configuration.

Parameters:

Name	Type	Description	Default
`password`	`str`	New password to set	required

Returns:

Type	Description
`bool`	True if successful

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

def set_password(self, password: str) -> bool:
    """Set the password in the account configuration.

    Args:
        password: New password to set

    Returns:
        True if successful
    """
    self.Config.password = password
    return True

set_username ¶

set_username(username: str) -> bool

Set the username in the account configuration.

Parameters:

Name	Type	Description	Default
`username`	`str`	New username to set	required

Returns:

Type	Description
`bool`	True if successful

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

def set_username(self, username: str) -> bool:
    """Set the username in the account configuration.

    Args:
        username: New username to set

    Returns:
        True if successful
    """
    self.Config.username = username
    return True

test_auths `async` ¶

test_auths(
    backup_auth: DomoAuth | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
) -> dict

Test both token and full authentication methods.

Attempts to validate both authentication methods and sets the best available as the target authentication.

Parameters:

Name	Type	Description	Default
`backup_auth`	`DomoAuth \| None`	Backup authentication to use if configured auths fail	`None`
`debug_api`	`bool`	Enable API debugging	`False`
`session`	`AsyncClient \| None`	HTTP client session (optional)	`None`

Returns:

Type	Description
`dict`	Dictionary with authentication test results

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

async def test_auths(
    self,
    backup_auth: DomoAuth | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
) -> dict:
    """Test both token and full authentication methods.

    Attempts to validate both authentication methods and sets the best available
    as the target authentication.

    Args:
        backup_auth: Backup authentication to use if configured auths fail
        debug_api: Enable API debugging
        session: HTTP client session (optional)

    Returns:
        Dictionary with authentication test results
    """
    ## test token auth
    try:
        await self.test_token_auth(debug_api=debug_api, session=session)

    except DomoError as e:
        print(f"testing token: {self.name}: {e}")

    ## test full auth
    try:
        await self.test_full_auth(debug_api=debug_api, session=session)

    except DomoError as e:
        print(f"testing full auth: {self.name}: {e}")

    ## generate target_auth
    try:
        self._set_target_auth(valid_backup_auth=backup_auth)
    except DACValidAuthError as e:
        print(f"{self.name}: unable to generate valid target_auth: {e}")
    return self.to_dict()

test_full_auth `async` ¶

test_full_auth(
    debug_api: bool = False,
    session: AsyncClient | None = None,
) -> bool

Test full authentication (username/password) for the account.

Generates a DomoFullAuth object and validates it against the target instance.

Parameters:

Name	Type	Description	Default
`debug_api`	`bool`	Enable API debugging	`False`
`session`	`AsyncClient \| None`	HTTP client session (optional)	`None`

Returns:

Type	Description
`bool`	True if authentication is valid, False otherwise

Raises:

Type	Description
`DAC_NoUserName`	If username is not configured
`DAC_NoPassword`	If password is not configured
`DAC_NoTargetInstance`	If target instance is not set

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

async def test_full_auth(
    self, debug_api: bool = False, session: httpx.AsyncClient | None = None
) -> bool:
    """Test full authentication (username/password) for the account.

    Generates a DomoFullAuth object and validates it against the target instance.

    Args:
        debug_api: Enable API debugging
        session: HTTP client session (optional)

    Returns:
        True if authentication is valid, False otherwise

    Raises:
        DAC_NoUserName: If username is not configured
        DAC_NoPassword: If password is not configured
        DAC_NoTargetInstance: If target instance is not set
    """
    self.is_valid_full_auth = False

    if not self.Config.username:
        raise DACNoUserNameError(self)

    if not self.Config.password:
        raise DACNoPasswordError(self)

    if not self.target_instance:
        raise DACNoTargetInstanceError(self)

    self._full_auth = DomoFullAuth(
        domo_instance=self.target_instance,
        domo_username=self.Config.username,
        domo_password=self.Config.password,
    )

    try:
        await self._full_auth.print_is_token(debug_api=debug_api, session=session)
        self.is_valid_full_auth = True

    except AuthError as e:
        dmcv.print_md(f"🤯 test_full_auth for: ***{self.name}*** returned {e}")

        self.is_valid_full_auth = False

    return self.is_valid_full_auth

test_token_auth `async` ¶

test_token_auth(
    debug_api: bool = False,
    session: AsyncClient | None = None,
) -> bool

Test token authentication for the account.

Generates a DomoTokenAuth object and validates it against the target instance.

Parameters:

Name	Type	Description	Default
`debug_api`	`bool`	Enable API debugging	`False`
`session`	`AsyncClient \| None`	HTTP client session (optional)	`None`

Returns:

Type	Description
`bool`	True if authentication is valid, False otherwise

Raises:

Type	Description
`DAC_NoAccessToken`	If access token is not configured
`DAC_NoTargetInstance`	If target instance is not set

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

async def test_token_auth(
    self, debug_api: bool = False, session: httpx.AsyncClient | None = None
) -> bool:
    """Test token authentication for the account.

    Generates a DomoTokenAuth object and validates it against the target instance.

    Args:
        debug_api: Enable API debugging
        session: HTTP client session (optional)

    Returns:
        True if authentication is valid, False otherwise

    Raises:
        DAC_NoAccessToken: If access token is not configured
        DAC_NoTargetInstance: If target instance is not set
    """

    self.is_valid_token_auth = False

    if not self.Config.domo_access_token:
        raise DACNoAccessTokenError(self)

    if not self.target_instance:
        raise DACNoTargetInstanceError(self)

    self._token_auth = DomoTokenAuth(
        domo_instance=self.target_instance,
        domo_access_token=self.Config.domo_access_token,
    )

    try:
        await self._token_auth.print_is_token(debug_api=debug_api, session=session)
        self.is_valid_token_auth = True
        self.target_auth = self._token_auth

    except AuthError as e:
        dmcv.print_md(f"🤯 test_token_auth for: ***{self.name}*** returned {e}")
        self.is_valid_token_auth = False

    return self.is_valid_token_auth

to_dict ¶

to_dict(return_snake_case: bool = False) -> dict

Convert credential information to dictionary.

Returns:

Type	Description
`dict`	Dictionary containing account ID, alias, instance, and auth validity status

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

def to_dict(self, return_snake_case: bool = False) -> dict:
    """Convert credential information to dictionary.

    Returns:
        Dictionary containing account ID, alias, instance, and auth validity status
    """

    s = super().to_dict(return_snake_case=return_snake_case)
    s.update(
        {
            "account_id": self.id,
            "alias": self.name,
            "target_instance": self.target_instance,
            "is_valid_full_auth": self.is_valid_full_auth,
            "is_valid_token_auth": self.is_valid_token_auth,
        }
    )
    return s

update_target_user_password `async` ¶

update_target_user_password(
    new_password: str,
    user_email: str | None = None,
    is_update_account: bool = True,
    target_auth: DomoAuth | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoAccountCredential

Update the password for the target user.

Parameters:

Name	Type	Description	Default
`new_password`	`str`	New password to set for the user	required
`user_email`	`str \| None`	Email address of user (defaults to configured username)	`None`
`is_update_account`	`bool`	Whether to update the account config with new password	`True`
`target_auth`	`DomoAuth \| None`	Authentication object (defaults to self.target_auth)	`None`
`debug_api`	`bool`	Enable API debugging	`False`
`session`	`AsyncClient \| None`	HTTP client session (optional)	`None`
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`
`**context_kwargs`		Additional context parameters	`{}`

Returns:

Type	Description
`DomoAccountCredential`	Self for method chaining

Raises:

Type	Description
`DACValidAuthError`	If target authentication is not available

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

async def update_target_user_password(
    self,
    new_password: str,
    user_email: str | None = None,
    is_update_account: bool = True,
    target_auth: DomoAuth | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> "DomoAccountCredential":
    """Update the password for the target user.

    Args:
        new_password: New password to set for the user
        user_email: Email address of user (defaults to configured username)
        is_update_account: Whether to update the account config with new password
        target_auth: Authentication object (defaults to self.target_auth)
        debug_api: Enable API debugging
        session: HTTP client session (optional)
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        Self for method chaining

    Raises:
        DACValidAuthError: If target authentication is not available
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    target_auth = target_auth or self.target_auth

    if not target_auth:
        raise DACValidAuthError(
            self,
            message="no target_auth, pass a valid backup_auth",
        )

    if not self.target_user:
        await self.get_target_user(
            context=context,
            user_email=user_email,
            target_auth=target_auth,
            **context_kwargs,
        )

    await self.target_user.reset_password(
        new_password=new_password,
        context=context,
        **context_kwargs,
    )

    self.set_password(new_password)

    if is_update_account:
        await self.update_config(context=context, **context_kwargs)

    return self

DomoAccount_Config `dataclass` ¶

DomoAccount_Config(
    data_provider_type: str,
    is_oauth: bool,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
)

Bases: DomoBase

DomoAccount Config abstract base class

from_dict `classmethod` ¶

from_dict(
    obj: dict[str, Any], parent: Any = None, **kwargs
)

Create instance from dict, handling camelCase, snake_case, and _field_map. Robust to missing or None _field_map. Uses _field_map for both directions.

Source code in src/crew_dcs/classes/DomoAccount/account_configs/_base.py

@classmethod
def from_dict(cls, obj: dict[str, Any], parent: Any = None, **kwargs):
    """
    Create instance from dict, handling camelCase, snake_case, and _field_map.
    Robust to missing or None _field_map. Uses _field_map for both directions.
    """
    # Get _field_map from dataclass field's default_factory if it exists
    field_map = {}
    if (
        hasattr(cls, "__dataclass_fields__")
        and "_field_map" in cls.__dataclass_fields__
    ):
        field_def = cls.__dataclass_fields__["_field_map"]
        if field_def.default_factory:
            field_map = field_def.default_factory()

    # Build a reverse map for input keys that are snake_case but need to be mapped to camelCase
    reverse_map = {
        v: k for k, v in field_map.items()
    }  # {"passphrase" : "passPhrase"}

    init_kwargs = {}

    init_kwargs.update(kwargs)

    for k, v in obj.items():
        if k in ["_search_metadata"]:
            continue

        # Try direct field_map, then reverse_map, then camel_to_snake
        if k in field_map:
            attr = field_map[k]

        elif k in reverse_map:
            attr = reverse_map[k]

        else:
            attr = cls._camel_to_snake(k)

        init_kwargs[attr] = v

    return cls(parent=parent, raw=obj, **init_kwargs)

to_dict ¶

to_dict(
    export_fields: list[str] | None = None,
    override_fn: Callable | None = None,
    return_snake_case: bool = False,
    **kwargs
) -> dict

Convert config to dictionary using provided fields or all _fields_for_serialization.

Source code in src/crew_dcs/classes/DomoAccount/account_configs/_base.py

def to_dict(
    self,
    export_fields: list[str] | None = None,
    override_fn: Callable | None = None,
    return_snake_case: bool = False,
    **kwargs,
) -> dict:
    """
    Convert config to dictionary using provided fields or all _fields_for_serialization.
    """
    result = {"allowExternalUse": self.allow_external_use}

    # Use all fields_for_serialization if not provided
    export_fields = [
        key
        for key in (
            export_fields
            or self._fields_for_serialization
            or self.__dataclass_fields__.keys()
        )
        if key
        not in [
            "raw",
            "_field_map",
            "_fields_for_serialization",
            "is_oauth",
            "data_provider_type",
            "__serialize_properties__",
        ]
    ]

    reverse_map = {v: k for k, v in self._field_map.items()}

    for attr in export_fields:
        val = getattr(self, attr, None)
        if val is not None:
            key = reverse_map.get(attr, self._snake_to_camel(attr))
            result[key] = val

    result.update(kwargs)

    if return_snake_case:
        result = {self._camel_to_snake(k): v for k, v in result.items()}

    if override_fn:
        result = override_fn(self)

    return result

validate_compare_serialization ¶

validate_compare_serialization(
    raise_on_mismatch: bool = True,
) -> bool

Validate that to_dict() output matches raw data keys (ignoring values).

Parameters:

Name	Type	Description	Default
`raise_on_mismatch`	`bool`	If True, raises AccountConfig_SerializationMismatchError on mismatch. If False, returns False on mismatch.	`True`

Returns:

Type	Description
`bool`	True if keys match, False otherwise (only if raise_on_mismatch=False)

Raises:

Type	Description
`AccountConfig_SerializationMismatchError`	If keys don't match and raise_on_mismatch=True

Source code in src/crew_dcs/classes/DomoAccount/account_configs/_base.py

def validate_compare_serialization(self, raise_on_mismatch: bool = True) -> bool:
    """
    Validate that to_dict() output matches raw data keys (ignoring values).

    Args:
        raise_on_mismatch: If True, raises AccountConfig_SerializationMismatchError on mismatch.
                         If False, returns False on mismatch.

    Returns:
        True if keys match, False otherwise (only if raise_on_mismatch=False)

    Raises:
        AccountConfig_SerializationMismatchError: If keys don't match and raise_on_mismatch=True
    """
    if not self.raw:
        return True

    to_dict_keys = set(self.to_dict(return_snake_case=False).keys())
    raw_keys = set(self.raw.keys()) - {"_search_metadata", "data_provider_type"}

    only_in_to_dict = to_dict_keys - raw_keys
    only_in_raw = raw_keys - to_dict_keys

    has_mismatch = bool(only_in_to_dict or only_in_raw)

    if has_mismatch and raise_on_mismatch:
        account_id = self.raw.get("_search_metadata", {}).get(
            "account_id", "unknown"
        )
        raise AccountConfig_SerializationMismatchError(
            cls_instance=self,
            data_provider_type=self.data_provider_type,
            account_id=str(account_id),
            only_in_to_dict=dict.fromkeys(only_in_to_dict),
            only_in_raw=dict.fromkeys(only_in_raw),
            value_mismatches={},
        )

    return not has_mismatch

DomoAccount_Default `dataclass` ¶

DomoAccount_Default(
    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: DomoEntity

display_url `property` ¶

display_url

returns the URL to the account in Domo

from_dict `classmethod` ¶

from_dict(
    auth: DomoAuth,
    obj: dict[str, Any],
    is_admin_summary: bool = True,
    new_cls: Any = None,
    **kwargs
)

converts data_v1_accounts API response into an accounts class object

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

@classmethod
def from_dict(
    cls,
    auth: DomoAuth,
    obj: dict[str, Any],
    is_admin_summary: bool = True,
    new_cls: Any = None,
    **kwargs,
):
    """converts data_v1_accounts API response into an accounts class object"""

    return new_cls(
        id=obj.get("id") or obj.get("databaseId"),
        name=obj.get("name"),
        display_name=obj.get("displayName"),
        data_provider_type=obj.get("dataProviderId") or obj.get("dataProviderType"),
        created_dt=cd.convert_epoch_millisecond_to_datetime(
            obj.get("createdAt") or obj.get("createDate")
        ),
        modified_dt=cd.convert_epoch_millisecond_to_datetime(
            obj.get("modifiedAt") or obj.get("lastModified")
        ),
        auth=auth,
        is_admin_summary=is_admin_summary,
        owners=obj.get("owners"),
        dataset_count=obj.get("datasetCount"),
        raw=obj,
        **kwargs,
    )

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,
    is_use_default_account_class=False,
    is_unmask=True,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

retrieves account metadata and attempts to retrieve config

Source code in src/crew_dcs/classes/DomoAccount/account_default.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,
    is_use_default_account_class=False,
    is_unmask=True,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """retrieves account metadata and attempts to retrieve config"""

    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_account_by_id(
        auth=auth,
        account_id=account_id,
        is_unmask=is_unmask,
        context=context,
    )

    if return_raw:
        return res

    obj = res.response

    acc = cls.from_dict(
        obj=obj,
        auth=auth,
        is_admin_summary=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

get_entity_by_id `async` `classmethod` ¶

get_entity_by_id(entity_id, **kwargs)

Alias for get_by_id

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

@classmethod
async def get_entity_by_id(cls, entity_id, **kwargs):
    """Alias for get_by_id"""
    return await cls.get_by_id(account_id=entity_id, **kwargs)

refresh `async` ¶

refresh(
    auth: DomoAuth = None,
    is_suppress_no_config: bool = True,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoAccount_Default

Reload entity properties from the API and refresh the account Config.

Thin wrapper that chains

super().refresh() — calls get_entity_by_id (i.e. get_by_id) to reload all entity-level attributes (name, data_provider_type, owners, etc.).
_get_config() — fetches and parses the account credential config.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Auth to use for config retrieval (defaults to `self.auth`).	`None`
`is_suppress_no_config`	`bool`	Suppress errors when no `DomoAccount_Config` subclass is registered for the account's `data_provider_type`. Defaults to `True` so callers can inspect `self.Config is None` rather than catching exceptions.	`True`
`debug_api`	`bool`	Enable API debug logging.	`False`
`session`	`AsyncClient \| None`	Optional reusable `httpx.AsyncClient`.	`None`
`context`	`RouteContext \| None`	Optional `RouteContext`.	`None`

Returns:

Type	Description
`DomoAccount_Default`	Self, mutated in-place with refreshed data.

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

async def refresh(
    self,
    auth: DomoAuth = None,
    is_suppress_no_config: bool = True,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> "DomoAccount_Default":
    """Reload entity properties from the API and refresh the account Config.

    Thin wrapper that chains:
      1. ``super().refresh()`` — calls ``get_entity_by_id`` (i.e. ``get_by_id``)
         to reload all entity-level attributes (name, data_provider_type, owners, etc.).
      2. ``_get_config()`` — fetches and parses the account credential config.

    Args:
        auth: Auth to use for config retrieval (defaults to ``self.auth``).
        is_suppress_no_config: Suppress errors when no ``DomoAccount_Config``
            subclass is registered for the account's ``data_provider_type``.
            Defaults to ``True`` so callers can inspect ``self.Config is None``
            rather than catching exceptions.
        debug_api: Enable API debug logging.
        session: Optional reusable ``httpx.AsyncClient``.
        context: Optional ``RouteContext``.

    Returns:
        Self, mutated in-place with refreshed data.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    await super().refresh(
        debug_api=debug_api,
        session=session,
        is_suppress_no_config=is_suppress_no_config,
    )

    await self._get_config(
        auth=auth or self.auth,
        is_suppress_no_config=is_suppress_no_config,
        debug_api=debug_api,
        session=session,
    )

    return self

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

Share this 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 \| AccountAccess_v1 \| str \| None`	Access level (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.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_default.py

async def share(
    self,
    user_id: int | None = None,
    group_id: int | None = None,
    access_level: AccountAccess | AccountAccess_v1 | str | None = None,
    session: httpx.AsyncClient | None = None,
    debug_api: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Share this 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 (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.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")

    # This method could not execute on any path before: it imported
    # `AccessLevel` from routes.account (the enum is `AccountAccess`) and
    # called `.generate_payload()`, which is defined nowhere in the library.
    # The real builders live in routes/account/access.py.
    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_payload(
        access_level=access_level,
        user_id=user_id,
        group_id=group_id,
    )

    res = await account_routes.share_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

upsert_target_account `async` ¶

upsert_target_account(
    target_auth: DomoAuth,
    account_name: str | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

upsert an account in a target instance with self.Config

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

async def upsert_target_account(
    self,
    target_auth: DomoAuth,  # valid auth for target destination
    account_name: str | None = None,  # defaults to self.display_name or self.name
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """
    upsert an account in a target instance with self.Config
    """
    from copy import deepcopy

    # Import here to avoid circular import
    from . import core

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

    account_name = account_name or self.display_name or self.name
    await logger.info(
        f"Upserting account {self.id} ({account_name}) to {target_auth.domo_instance}"
    )

    return await core.DomoAccounts.upsert_account(
        auth=target_auth,
        account_name=account_name,
        account_config=deepcopy(self.Config),
        data_provider_type=self.data_provider_type,
        context=context,
    )

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

DomoAccounts `dataclass` ¶

DomoAccounts(
    auth: DomoAuth,
    accounts: list[DomoAccount] = None,
    oauths: list[DomoAccount_OAuth] = None,
)

Bases: DomoManager

get_accounts_queryapi `async` ¶

get_accounts_queryapi(
    additional_filters_ls=None,
    is_use_default_account_class: bool = False,
    return_raw: bool = False,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: AsyncClient = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

v2 api for works with group_account_v2 beta

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

async def get_accounts_queryapi(
    self,
    additional_filters_ls=None,
    is_use_default_account_class: bool = False,
    return_raw: bool = False,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: httpx.AsyncClient = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """v2 api for works with group_account_v2 beta"""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop - 1,
        **context_kwargs,
    )

    res = await datacenter_routes.search_datacenter(
        auth=self.auth,
        entity_type=datacenter_routes.Datacenter_Enum.ACCOUNT.value,
        additional_filters_ls=additional_filters_ls,
        context=context,
    )

    if return_raw:
        return res

    if len(res.response) == 0:
        self.accounts = []
        return self.accounts

    self.accounts = [
        DomoAccount.from_dict(
            auth=self.auth,
            obj=account_obj,
            is_use_default_class=is_use_default_account_class,
        )
        for account_obj in res.response
    ]
    return self.accounts

search_by_name `async` ¶

search_by_name(
    account_name: str,
    data_provider_type: str | None = None,
    is_use_default_account_class: bool = True,
    is_suppress_not_found_exception: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoAccount | None

Search for an account by name (matches display_name or name).

Parameters:

Name	Type	Description	Default
`account_name`	`str`	Account name to search for (case-insensitive)	required
`data_provider_type`	`str \| None`	Optional filter by data provider type	`None`
`is_use_default_account_class`	`bool`	Use default account class	`True`
`is_suppress_not_found_exception`	`bool`	Suppress not found exceptions	`False`
`context`	`RouteContext \| None`	Optional RouteContext for API call configuration	`None`
`**context_kwargs`		Additional context parameters (session, debug_api, etc.)	`{}`

Returns:

Type	Description
`DomoAccount \| None`	Matching DomoAccount or None if not found

Raises:

Type	Description
`DomoError`	If account retrieval fails

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

async def search_by_name(
    self,
    account_name: str,
    data_provider_type: str | None = None,
    is_use_default_account_class: bool = True,
    is_suppress_not_found_exception: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoAccount | None:
    """Search for an account by name (matches display_name or name).

    Args:
        account_name: Account name to search for (case-insensitive)
        data_provider_type: Optional filter by data provider type
        is_use_default_account_class: Use default account class
        is_suppress_not_found_exception: Suppress not found exceptions
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        Matching DomoAccount or None if not found

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

    await self.get(
        context=context,
        is_use_default_account_class=is_use_default_account_class,
    )

    for account in self.accounts:
        # Check both name and display_name for matching
        name_match = False
        if (
            account.display_name
            and account.display_name.lower() == account_name.lower()
        ) or (account.name and account.name.lower() == account_name.lower()):
            name_match = True

        if not name_match:
            continue

        # If data_provider_type specified, must match
        if data_provider_type and data_provider_type != account.data_provider_type:
            continue

        return account

    if is_suppress_not_found_exception:
        return None

    raise DomoAccounts_NoAccount(
        cls=self.__class__,
        message=f"No account found with name '{account_name}'",
        domo_instance=self.auth.domo_instance,
    )

upsert_account `async` `classmethod` ¶

upsert_account(
    auth: DomoAuth,
    account_id: str | None = None,
    account_name: str | None = None,
    account_config: AccountConfig = None,
    data_provider_type: str | None = None,
    debug_api: bool = False,
    return_raw: bool = False,
    return_search: bool = False,
    is_use_default_account_class: bool = True,
    session: AsyncClient = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

search for an account and upsert it

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

@classmethod
async def upsert_account(
    cls,
    auth: DomoAuth,
    account_id: str | None = None,
    account_name: str | None = None,
    account_config: AccountConfig = None,
    data_provider_type: str | None = None,
    debug_api: bool = False,
    return_raw: bool = False,
    return_search: bool = False,
    is_use_default_account_class: bool = True,
    session: httpx.AsyncClient = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """search for an account and upsert it"""

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

    if not account_name and not account_id:
        raise UpsertAccount_MatchCriteriaError(domo_instance=auth.domo_instance)

    data_provider_type = data_provider_type or (
        account_config and account_config.data_provider_type
    )
    acc = None

    if account_id:
        with contextlib.suppress(dmde.DomoError):
            acc = await DomoAccount.get_by_id(
                auth=auth,
                account_id=account_id,
                context=context,
                is_use_default_account_class=is_use_default_account_class,
            )

    if account_name and not acc:
        try:
            domo_accounts = DomoAccounts(auth=auth)
            acc = await domo_accounts.search_by_name(
                account_name=account_name,
                data_provider_type=data_provider_type,
                is_use_default_account_class=is_use_default_account_class,
                context=context,
            )
        except dmde.DomoError:
            pass

    if return_search:
        return acc

    if not isinstance(
        acc, DomoAccount_Default | DomoAccount | DomoAccountCredential
    ):
        await logger.info(
            f"Creating account {account_name} in {auth.domo_instance}"
        )

        return await DomoAccount.create_account(
            account_name=account_name,
            config=account_config,
            auth=auth,
            context=context,
            return_raw=return_raw,
        )

    if account_name and account_id:
        await logger.info(
            f"Updating account {acc.id} - {acc.display_name or acc.name} in {auth.domo_instance}"
        )

        await acc.update_name(
            account_name=account_name,
            context=context,
            return_raw=return_raw,
        )

    if account_config:  # upsert account
        acc.Config = account_config

        await logger.info(f"Updating config for account {acc.id}")

        await acc.update_config(context=context, return_raw=return_raw)

    return acc

DomoAccount

DomoAccount ¶

DomoAccount dataclass ¶

from_dict classmethod ¶

DomoAccountCredential dataclass ¶

from_dict classmethod ¶

get_target_access_token async ¶

get_target_user async ¶

regenerate_target_access_token async ¶

set_access_token ¶

set_password ¶

set_username ¶

test_auths async ¶

test_full_auth async ¶

test_token_auth async ¶

to_dict ¶

update_target_user_password async ¶

DomoAccount_Config dataclass ¶

from_dict classmethod ¶

to_dict ¶

validate_compare_serialization ¶

DomoAccount_Default dataclass ¶

display_url property ¶

from_dict classmethod ¶

get_by_id async classmethod ¶

get_entity_by_id async classmethod ¶

refresh async ¶

share async ¶

upsert_target_account async ¶

DomoAccount_OAuth dataclass ¶

create async classmethod ¶

delete async ¶

get_by_id async classmethod ¶

share async ¶

update_config async ¶

update_name async ¶

DomoAccounts dataclass ¶

get_accounts_queryapi async ¶

search_by_name async ¶

upsert_account async classmethod ¶

Modules¶

DomoAccount `dataclass` ¶

from_dict `classmethod` ¶

DomoAccountCredential `dataclass` ¶

from_dict `classmethod` ¶

get_target_access_token `async` ¶

get_target_user `async` ¶

regenerate_target_access_token `async` ¶

test_auths `async` ¶

test_full_auth `async` ¶

test_token_auth `async` ¶

update_target_user_password `async` ¶

DomoAccount_Config `dataclass` ¶

from_dict `classmethod` ¶

DomoAccount_Default `dataclass` ¶

display_url `property` ¶

from_dict `classmethod` ¶

get_by_id `async` `classmethod` ¶

get_entity_by_id `async` `classmethod` ¶

refresh `async` ¶

share `async` ¶

upsert_target_account `async` ¶

DomoAccount_OAuth `dataclass` ¶

create `async` `classmethod` ¶

delete `async` ¶

get_by_id `async` `classmethod` ¶

share `async` ¶

update_config `async` ¶

update_name `async` ¶

DomoAccounts `dataclass` ¶

get_accounts_queryapi `async` ¶

search_by_name `async` ¶

upsert_account `async` `classmethod` ¶