auth_utils

auth_utils

get_auth_from_codeengine async

get_auth_from_codeengine(
    config_auth: DomoAuth,
    sudo_package_id: str,
    sudo_package_version: str,
    retrieval_account_name: str,
    is_return_access_token: bool = True,
    target_instance: str | None = None,
    function_name: str = "get_account",
    debug_api: bool = False,
    return_raw: bool = False,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoTokenAuth | ResponseGetData

Retrieve account object in clear text using a CodeEngine package. optionally converts to DomoTokenAuth.

Parameters:

Name	Type	Description	Default
config_auth	DomoAuth	DomoAuth with access to the Domo instance (authenticates the API request)	required
target_instance	str | None	The Domo instance for which credentials are requested	None
sudo_package_id	str	The CodeEngine package ID	required
sudo_package_version	str	The version of the CodeEngine package	required
retreival_account_name		The alias of the account to retrieve (must be shared with the package)	required
function_name	str	The CodeEngine function to call (default: "get_account")	'get_account'
is_return_access_token	bool	If True, returns a DomoTokenAuth; if False, returns the raw ResponseGetData	True

Returns:

Type	Description
DomoTokenAuth | ResponseGetData	DomoTokenAuth for the requested account, or ResponseGetData if is_return_access_token is False

Raises:

Type	Description
AssertionError	If the returned token is invalid

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

@log_call(level_name="integration", color="cyan", log_level="info")
async def get_auth_from_codeengine(
    config_auth: dmda.DomoAuth,
    sudo_package_id: str,
    sudo_package_version: str,
    retrieval_account_name: str,
    # for returing DomoTokenAuth
    is_return_access_token: bool = True,
    target_instance: str | None = None,  # required for generating a DomoTokenAuth
    function_name: str = "get_account",  # code engine function name
    debug_api: bool = False,
    return_raw: bool = False,
    context: RouteContext | None = None,
    **context_kwargs,
) -> dmda.DomoTokenAuth | ResponseGetData:
    """
    Retrieve account object in clear text using a CodeEngine package.
    optionally converts to DomoTokenAuth.

    Args:
        config_auth: DomoAuth with access to the Domo instance (authenticates the API request)
        target_instance: The Domo instance for which credentials are requested
        sudo_package_id: The CodeEngine package ID
        sudo_package_version: The version of the CodeEngine package
        retreival_account_name: The alias of the account to retrieve (must be shared with the package)
        function_name: The CodeEngine function to call (default: "get_account")
        is_return_access_token: If True, returns a DomoTokenAuth; if False, returns the raw ResponseGetData

    Returns:
        DomoTokenAuth for the requested account, or ResponseGetData if is_return_access_token is False

    Raises:
        AssertionError: If the returned token is invalid
    """

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

    res = await ce_routes.execute_codeengine_function(
        auth=config_auth,
        package_id=sudo_package_id,
        version=sudo_package_version,
        function_name=function_name,
        input_variables={"auth_name": retrieval_account_name},
        debug_api=debug_api,
        return_raw=return_raw,
        context=context,
    )

    if return_raw:
        return res

    if not is_return_access_token:
        await logger.info(
            f"Retrieved auth response for account '{retrieval_account_name}' from Code Engine - returning raw response"
        )
        return res

    await logger.info(
        f"Retrieved auth response for account '{retrieval_account_name}' from Code Engine - generating DomoTokenAuth"
    )
    target_auth = dmda.DomoTokenAuth(
        domo_access_token=res.response["properties"]["domoAccessToken"],
        domo_instance=target_instance,
    )
    try:
        await logger.info(
            f"Validating retrieved DomoTokenAuth - {retrieval_account_name}"
        )
        assert await target_auth.print_is_token()

    except (AssertionError, DomoError) as e:
        message = f"Invalid auth retrieved from CodeEngine for account '{retrieval_account_name}' - {e}"
        await logger.error(message)
        raise AssertionError(message)  # noqa: B904

    return target_auth