core

core

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

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