DeployAccount

DeployAccount

DeployAccount dataclass

DeployAccount(
    auth: DomoAuth,
    creds: dict[str, Any],
    config_types: Sequence[ConfigTypeSpecifier],
    account_name: str,
    description: str | None = None,
    is_available_as_unencrypted: bool = True,
    context: RouteContext | None = None,
)

Deploy multiple DomoAccount objects that share a credential bundle.

Attributes:

Name	Type	Description
auth	DomoAuth	Auth used to create / upsert the accounts.
creds	dict[str, Any]	Shared credential dictionary. Keys should match the dataclass field names on the target DomoAccount_Config subclasses (e.g. account, username, private_key, access_key).
config_types	Sequence[ConfigTypeSpecifier]	Collection of config type specifiers. Each entry may be: an AccountConfig enum member a concrete DomoAccount_Config subclass a data_provider_type string (e.g. "snowflake-key-pair-unload-v2")
account_name	str	Base account name to use. A short provider suffix is appended to make each account name unique per config type.
description	str | None	Optional human friendly description of the bundle. Not sent to the Domo APIs; intended for logging / debugging.
is_available_as_unencrypted	bool	Controls the allow_external_use flag on each generated config. When False we still build configs normally, but mark them as not available for external unencrypted use.

deploy async

deploy(
    *,
    debug_api: bool = False,
    debug_prn: bool = False,
    dry_run: bool = False
) -> list[DomoAccount]

Create or update one DomoAccount per config type.

Returns:

Type	Description
list[DomoAccount]	List of created / upserted DomoAccount instances in the same
list[DomoAccount]	order as config_types.

Source code in src/crew_dcs/integrations/DeployAccount.py

@log_call(level_name="integration", color="cyan", log_level="info")
async def deploy(
    self,
    *,
    debug_api: bool = False,
    debug_prn: bool = False,
    dry_run: bool = False,
) -> list[DomoAccount]:
    """Create or update one `DomoAccount` per config type.

    Returns:
        List of created / upserted `DomoAccount` instances in the same
        order as `config_types`.
    """

    # Build effective route context (merge existing context + overrides)
    ctx = RouteContext.build_context(
        self.context,
        debug_api=debug_api,
        dry_run=dry_run,
    )
    # If either flag says dry_run, treat it as a dry run
    dry_run = dry_run or ctx.dry_run

    accounts: list[DomoAccount] = []

    await logger.info(
        "Starting DeployAccount deployment",
        method="COMMENT",
        account_name=self.account_name,
        description=self.description,
        is_available_as_unencrypted=self.is_available_as_unencrypted,
        num_config_types=len(self.config_types),
        dry_run=dry_run,
        context_has_session=ctx.session is not None,
        context_debug_api=ctx.debug_api,
    )

    for spec in self.config_types:
        config_cls, provider_type = self._resolve_config_class(spec)
        config = self._build_config(config_cls)

        # Mirror the per-config allow_external_use flag against the
        # DeployAccount's high‑level `is_available_as_unencrypted` flag.  # noqa: RUF003
        config.allow_external_use = self.is_available_as_unencrypted

        account_display_name = self._build_account_name(
            self.account_name, provider_type
        )

        if dry_run:
            await logger.info(
                "[DRY RUN] Would deploy DomoAccount",
                method="COMMENT",
                account_name=account_display_name,
                provider_type=provider_type,
                config_class=config_cls.__name__,
                creds_keys=sorted(self.creds.keys()),
                session_has=ctx.session is not None,
                context_debug_api=ctx.debug_api,
            )
            continue

        acc = await DomoAccounts.upsert_account(
            auth=self.auth,
            account_name=account_display_name,
            account_config=config,
            data_provider_type=provider_type,
            debug_api=ctx.debug_api,
            debug_prn=debug_prn,
            session=ctx.session,
        )

        accounts.append(acc)

        await logger.info(
            "Deployed DomoAccount",
            method="COMMENT",
            account_id=getattr(acc, "id", None),
            account_name=account_display_name,
            provider_type=provider_type,
        )

    return accounts