providers

providers

Built-in authentication providers.

This module contains the standard authentication providers for environment variables, CodeEngine sudo, and OnePassword.

CodeEngineAuthProvider

CodeEngineAuthProvider()

Provider for CodeEngine sudo authentication.

Source code in src/crew_dcs/integrations/auth/providers.py

def __init__(self):
    self.method = AuthMethod.CODEENGINE

try_auth async

try_auth(
    sudo_function_instance_auth: DomoAuth | None = None,
    target_instance: str | None = None,
    account_name: str | None = None,
    sudo_package_id: str | None = None,
    sudo_package_version: str | None = None,
    function_name: str = "get_account",
    debug_api: bool = False,
    **kwargs
) -> DomoAuth | None

Try to authenticate via CodeEngine sudo.

Parameters:

Name	Type	Description	Default
sudo_function_instance_auth	DomoAuth | None	Authenticated auth for CodeEngine calls	None
target_instance	str | None	Target Domo instance	None
account_name	str | None	Account identifier (defaults to sdk_{target_instance})	None
sudo_package_id	str | None	CodeEngine package ID	None
sudo_package_version	str | None	CodeEngine package version	None
function_name	str	CodeEngine function name	'get_account'
debug_api	bool	Enable debug output	False
**kwargs		Additional arguments (ignored)	{}

Returns:

Type	Description
DomoAuth | None	DomoAuth if successful, None if prerequisites missing or auth fails

Source code in src/crew_dcs/integrations/auth/providers.py

async def try_auth(
    self,
    sudo_function_instance_auth: dmda.DomoAuth | None = None,
    target_instance: str | None = None,
    account_name: str | None = None,
    sudo_package_id: str | None = None,
    sudo_package_version: str | None = None,
    function_name: str = "get_account",
    debug_api: bool = False,
    **kwargs,
) -> dmda.DomoAuth | None:
    """
    Try to authenticate via CodeEngine sudo.

    Args:
        sudo_function_instance_auth: Authenticated auth for CodeEngine calls
        target_instance: Target Domo instance
        account_name: Account identifier (defaults to sdk_{target_instance})
        sudo_package_id: CodeEngine package ID
        sudo_package_version: CodeEngine package version
        function_name: CodeEngine function name
        debug_api: Enable debug output
        **kwargs: Additional arguments (ignored)

    Returns:
        DomoAuth if successful, None if prerequisites missing or auth fails
    """
    from crew_dcs.integrations.auth_utils import get_auth_from_codeengine

    if (
        not sudo_function_instance_auth
        or not sudo_package_id
        or not sudo_package_version
    ):
        await logger.error("CodeEngine auth prerequisites not met")
        raise ValueError("Missing prerequisites for CodeEngine auth")

    if not target_instance:
        await logger.error("CodeEngine auth requires target_instance")
        raise ValueError("Missing target_instance for CodeEngine auth")

    await logger.info(
        f"Attempting CodeEngine auth for {target_instance} "
        f"(package: {sudo_package_id})"
    )

    try:
        return await get_auth_from_codeengine(
            config_auth=sudo_function_instance_auth,
            target_instance=target_instance,
            sudo_package_id=sudo_package_id,
            sudo_package_version=sudo_package_version,
            retrieval_account_name=account_name or f"sdk_{target_instance}",
            function_name=function_name,
            debug_api=debug_api,
        )
    except (KeyError, AssertionError) as e:
        # Expected failures: missing account, invalid token
        await logger.error(f"CodeEngine auth failed: {e}")
        raise e

EnvironmentAuthProvider

EnvironmentAuthProvider()

Provider for environment variable authentication.

Source code in src/crew_dcs/integrations/auth/providers.py

def __init__(self):
    self.method = AuthMethod.ENVIRONMENT

try_auth async

try_auth(
    target_instance: str | None = None,
    domo_instance_env_var: str = "DOMO_INSTANCE",
    domo_token_env_var: str = "DOMO_ACCESS_TOKEN",
    **kwargs
) -> DomoAuth | None

Try to authenticate from environment variables.

Parameters:

Name	Type	Description	Default
target_instance	str | None	Override instance name (sets env var temporarily)	None
domo_instance_env_var	str	Environment variable name for instance	'DOMO_INSTANCE'
domo_token_env_var	str	Environment variable name for token	'DOMO_ACCESS_TOKEN'
**kwargs		Additional arguments (ignored)	{}

Returns:

Type	Description
DomoAuth | None	DomoAuth if successful, None if env vars not available

Source code in src/crew_dcs/integrations/auth/providers.py

async def try_auth(
    self,
    target_instance: str | None = None,
    domo_instance_env_var: str = "DOMO_INSTANCE",
    domo_token_env_var: str = "DOMO_ACCESS_TOKEN",
    **kwargs,
) -> dmda.DomoAuth | None:
    """
    Try to authenticate from environment variables.

    Args:
        target_instance: Override instance name (sets env var temporarily)
        domo_instance_env_var: Environment variable name for instance
        domo_token_env_var: Environment variable name for token
        **kwargs: Additional arguments (ignored)

    Returns:
        DomoAuth if successful, None if env vars not available
    """
    await logger.debug(
        f"Attempting environment auth for {target_instance or 'default instance'}"
    )

    try:
        from crew_dcs.integrations.auth_utils import get_auth_from_env

        # Temporarily override env var if target_instance provided
        original = None
        if target_instance:
            original = os.environ.get(domo_instance_env_var)
            os.environ[domo_instance_env_var] = target_instance

        auth = await get_auth_from_env(
            domo_instance_env_var=domo_instance_env_var,
            domo_token_env_var=domo_token_env_var,
        )

        # Restore original env var
        if target_instance:
            if original:
                os.environ[domo_instance_env_var] = original
            else:
                os.environ.pop(domo_instance_env_var, None)

        return auth

    except (KeyError, AssertionError) as e:
        await logger.debug(f"Environment auth failed: {e}")
        raise e

OnePasswordAuthProvider

OnePasswordAuthProvider()

Provider for 1Password authentication.

Source code in src/crew_dcs/integrations/auth/providers.py

def __init__(self):
    self.method = AuthMethod.ONEPASSWORD

try_auth async

try_auth(
    vault_id: str | None = None,
    item_title: str | None = None,
    target_instance: str | None = None,
    domo_instance: str | None = None,
    client=None,
    debug_api: bool = False,
    **kwargs
) -> DomoAuth | None

Try to authenticate via 1Password.

Parameters:

Name	Type	Description	Default
vault_id	str | None	1Password vault ID	None
item_title	str | None	1Password item title (defaults to sdk_{target_instance})	None
target_instance	str | None	Target Domo instance name	None
domo_instance	str | None	Override for final instance name	None
client		Optional existing 1Password client	None
debug_api	bool	Enable debug output	False
**kwargs		Additional arguments (ignored)	{}

Returns:

Type	Description
DomoAuth | None	DomoAuth if successful, None if prerequisites missing or auth fails

Source code in src/crew_dcs/integrations/auth/providers.py

async def try_auth(
    self,
    vault_id: str | None = None,
    item_title: str | None = None,
    target_instance: str | None = None,
    domo_instance: str | None = None,
    client=None,
    debug_api: bool = False,
    **kwargs,
) -> dmda.DomoAuth | None:
    """
    Try to authenticate via 1Password.

    Args:
        vault_id: 1Password vault ID
        item_title: 1Password item title (defaults to sdk_{target_instance})
        target_instance: Target Domo instance name
        domo_instance: Override for final instance name
        client: Optional existing 1Password client
        debug_api: Enable debug output
        **kwargs: Additional arguments (ignored)

    Returns:
        DomoAuth if successful, None if prerequisites missing or auth fails
    """

    from .onepassword import get_auth_from_onepass

    if not vault_id:
        await logger.error("OnePassword auth requires vault_id")
        raise ValueError("vault_id is required for OnePassword auth")

    # Determine item title and instance
    resolved_item_title = item_title or (
        target_instance and f"sdk_{target_instance}"
    )
    resolved_instance = domo_instance or target_instance

    if not resolved_item_title:
        await logger.error(
            "OnePassword auth requires item_title or target_instance"
        )
        raise ValueError(
            "item_title or target_instance is required for OnePassword auth"
        )

    await logger.info(f"Attempting OnePassword auth for {resolved_item_title}")

    return await get_auth_from_onepass(
        vault_id=vault_id,
        item_title=resolved_item_title,
        domo_instance=resolved_instance,
        client=client,
        debug_api=debug_api,
    )