account_default

account_default

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

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.

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 async

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