auth

auth ¶

Domo authentication classes for various authentication methods.

This module re-exports authentication classes for different Domo auth methods: - DomoAuth: Base authentication class - DomoFullAuth: Username/password authentication - DomoTokenAuth: Access token authentication - DomoDeveloperAuth: OAuth2 client credentials authentication - DomoJupyterAuth: Base Jupyter authentication - DomoJupyterFullAuth: Jupyter with username/password - DomoJupyterTokenAuth: Jupyter with access token

Type Hinting: All authentication classes (DomoTokenAuth, DomoFullAuth, DomoDeveloperAuth, etc.) inherit from DomoAuth and implement the same interface via shared mixins, so functions can accept DomoAuth and work with any auth subclass.

DomoAuth ¶

Bases: ABC

Abstract contract for Domo authentication implementations.

All concrete auth classes must inherit from this class and provide the required authentication surface used by routes and entities.

auth_header `abstractmethod` `property` ¶

auth_header: dict

Generate authentication headers for API requests.

url_manual_login `property` ¶

url_manual_login: str

Generate the manual login URL for the Domo instance.

cache_stats ¶

cache_stats() -> dict

Get HTTP cache statistics.

Source code in src/crew_dcs/auth/base.py

def cache_stats(self) -> dict:
    """Get HTTP cache statistics."""
    if hasattr(self, "_cached_session") and self._cached_session:
        transport = getattr(self._cached_session, "_transport", None)
        if transport and hasattr(transport, "get_stats"):
            return transport.get_stats()

    return {
        "request_cache": {
            "total_entries": 0,
            "hits": 0,
            "misses": 0,
            "hit_rate": 0.0,
        },
        "collection_cache": {
            "total_entries": 0,
            "hits": 0,
            "misses": 0,
            "hit_rate": 0.0,
        },
    }

clear_cache ¶

clear_cache() -> None

Clear all HTTP caches for this auth instance.

Source code in src/crew_dcs/auth/base.py

def clear_cache(self) -> None:
    """Clear all HTTP caches for this auth instance."""
    if hasattr(self, "_cached_session") and self._cached_session:
        transport = getattr(self._cached_session, "_transport", None)
        if transport and hasattr(transport, "clear_cache"):
            transport.clear_cache()

close_session `async` ¶

close_session() -> None

Close and drop the pooled session cached on this auth, if any.

A reused session is otherwise held open until the auth is GC'd. Call this (or use the auth as an async context manager) for clean teardown::

async with DomoTokenAuth(...) as auth:
    await DomoCard.create(auth=auth, ...)   # reuses one pooled session
# session closed here

Source code in src/crew_dcs/auth/base.py

async def close_session(self) -> None:
    """Close and drop the pooled session cached on this auth, if any.

    A reused session is otherwise held open until the auth is GC'd. Call this
    (or use the auth as an async context manager) for clean teardown::

        async with DomoTokenAuth(...) as auth:
            await DomoCard.create(auth=auth, ...)   # reuses one pooled session
        # session closed here
    """
    session = getattr(self, "_cached_session", None)
    if session is not None:
        try:
            if not getattr(session, "is_closed", False):
                await session.aclose()
        finally:
            self._cached_session = None

elevate_otp `async` ¶

elevate_otp(
    one_time_password: str,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
) -> ResponseGetData

Elevate the authentication to include OTP (One-Time Password) if required.

Source code in src/crew_dcs/auth/base.py

@log_call(
    level_name="auth",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def elevate_otp(
    self,
    one_time_password: str,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
) -> ResponseGetData:
    """Elevate the authentication to include OTP (One-Time Password) if required."""
    if session is None:
        async with httpx.AsyncClient() as client_session:
            from ..routes import auth as auth_routes

            return await auth_routes.elevate_user_otp(
                auth=self,
                debug_api=debug_api,
                session=client_session,
                one_time_password=one_time_password,
                debug_num_stacks_to_drop=debug_num_stacks_to_drop,
            )
    from ..routes import auth as auth_routes

    return await auth_routes.elevate_user_otp(
        auth=self,
        debug_api=debug_api,
        session=session,
        one_time_password=one_time_password,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
    )

get_auth_token `abstractmethod` `async` ¶

get_auth_token(**context_kwargs) -> str | ResponseGetData

Retrieve or generate an authentication token.

Source code in src/crew_dcs/auth/base.py

@abstractmethod
async def get_auth_token(
    self,
    **context_kwargs,
) -> str | ResponseGetData:
    """Retrieve or generate an authentication token."""
    raise NotImplementedError("Subclasses must implement get_auth_token.")

invalidate_cache `async` ¶

invalidate_cache(url_pattern: str) -> None

Manually invalidate cache entries matching URL pattern.

Source code in src/crew_dcs/auth/base.py

async def invalidate_cache(self, url_pattern: str) -> None:
    """Manually invalidate cache entries matching URL pattern."""
    if hasattr(self, "_cached_session") and self._cached_session:
        transport = getattr(self._cached_session, "_transport", None)
        if transport and hasattr(transport, "_invalidate_by_url"):
            await transport._invalidate_by_url(url_pattern)

print_is_token `async` ¶

print_is_token(
    token_name: str | None = None, **context_kwargs
) -> bool

Print token status and return True if token is valid, otherwise False.

Source code in src/crew_dcs/auth/base.py

async def print_is_token(
    self,
    token_name: str | None = None,
    **context_kwargs,
) -> bool:
    """Print token status and return True if token is valid, otherwise False."""
    self.token_name = token_name or self.token_name

    if not self.token:
        await self.get_auth_token(**context_kwargs)

    if not self.is_valid_token:
        await self.who_am_i(**context_kwargs)

    if not self.is_valid_token:
        print(
            f"🚧 failed to retrieve {self.token_name} token from {self.domo_instance}"
        )
        return False

    try:
        print(f"🎉 {self.token_name} token retrieved from {self.domo_instance} ⚙️")
    except UnicodeEncodeError:
        print(
            f"[SUCCESS] {self.token_name} token retrieved from {self.domo_instance}"
        )
    return True

set_cache_strategy ¶

set_cache_strategy(strategy) -> None

Change cache invalidation strategy.

Source code in src/crew_dcs/auth/base.py

def set_cache_strategy(self, strategy) -> None:
    """Change cache invalidation strategy."""
    if hasattr(self, "_cached_session") and self._cached_session:
        transport = getattr(self._cached_session, "_transport", None)
        if transport and hasattr(transport, "invalidator"):
            transport.invalidator.strategy = strategy

who_am_i `async` ¶

who_am_i(**context_kwargs) -> ResponseGetData

Perform an API call to identify the user associated with the token.

Source code in src/crew_dcs/auth/base.py

@log_call(
    level_name="auth",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
    color="cyan",
)
async def who_am_i(
    self,
    **context_kwargs,
) -> ResponseGetData:
    """Perform an API call to identify the user associated with the token."""
    from ..routes import auth as auth_routes

    res = await auth_routes.who_am_i(
        auth=self,
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )

    if res.is_success:
        self.is_valid_token = True

    if res.response and isinstance(res.response, dict):
        self.user_id = res.response.get("id")

    return res

DomoDeveloperAuth `dataclass` ¶

DomoDeveloperAuth(
    domo_client_id: str,
    domo_client_secret: str,
    domo_instance: str,
    token_name: str | None = None,
    token: str | None = None,
    user_id: str | None = None,
    is_valid_token: bool = False,
    cache_responses: bool = False,
    cache_size: int = 1000,
    cache_default_ttl: int = 300,
    cache_strategy: str | None = None,
    custom_cache_rules: dict[str, list[str]] | None = None,
)

Bases: DomoAuth

Developer authentication using client credentials.

This authentication method uses OAuth2 client credentials (client ID and secret) to obtain bearer tokens. This is typically used for applications built on Domo's developer platform and requires developer app registration.

Attributes:

Name	Type	Description
`domo_client_id`	`str`	OAuth2 client ID from developer app registration
`domo_client_secret`	`str`	OAuth2 client secret (not shown in repr)
`domo_instance`	`str`	The Domo instance identifier
`token_name`	`str \| None`	Name identifier for the token
`token`	`str \| None`	The bearer token (not shown in repr)
`user_id`	`str \| None`	The authenticated user's ID
`is_valid_token`	`bool`	Whether the current token is valid
`cache_responses`	`bool`	Enable HTTP response caching (default: False)
`cache_size`	`int`	Maximum number of cache entries (default: 1000)
`cache_default_ttl`	`int`	Default cache TTL in seconds (default: 300)
`cache_strategy`	`str \| None`	Cache invalidation strategy (default: None, uses SMART)
`custom_cache_rules`	`dict \| None`	Custom cache invalidation rules (default: None)

Example

auth = DomoDeveloperAuth( ... domo_client_id="your-client-id", ... domo_client_secret="", # pragma: allowlist secret ... domo_instance="mycompany" ... ) token = await auth.get_auth_token()

auth_header `property` ¶

auth_header: dict

Generate the authentication header for developer token authentication.

get_auth_token `async` ¶

get_auth_token(**context_kwargs) -> str

Retrieve the developer token using client credentials and update internal attributes.

Source code in src/crew_dcs/auth/developer.py

@log_call(
    level_name="auth",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_auth_token(
    self,
    **context_kwargs,
) -> str:
    """Retrieve the developer token using client credentials and update internal attributes."""
    from ..routes import auth as auth_routes

    res = await auth_routes.get_developer_auth(
        auth=None,
        domo_client_id=self.domo_client_id,
        domo_client_secret=self.domo_client_secret,
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )

    if isinstance(res, ResponseGetData) and res.is_success and res.response:
        self.is_valid_token = True
        self.token = str(
            res.response.get("access_token", "")
            if isinstance(res.response, dict)
            else ""
        )
        self.user_id = (
            res.response.get("userId") if isinstance(res.response, dict) else ""
        )
        if isinstance(self.user_id, int):
            self.user_id = str(self.user_id)
        self.domo_instance = (
            res.response.get("domain", self.domo_instance)
            if isinstance(res.response, dict)
            else ""
        )
        self.token_name = self.token_name or "developer_auth"
        return self.token

    raise AuthError(message="Failed to retrieve developer token")

who_am_i `async` ¶

who_am_i(**context_kwargs) -> ResponseGetData

Use cached user_id from get_auth_token response (no separate API call).

Source code in src/crew_dcs/auth/developer.py

async def who_am_i(
    self,
    **context_kwargs,
) -> ResponseGetData:
    """Use cached user_id from get_auth_token response (no separate API call)."""
    if not self.user_id:
        await self.get_auth_token(**context_kwargs)
    return ResponseGetData(
        status=200,
        response={"id": self.user_id},
        is_success=True,
    )

DomoFullAuth `dataclass` ¶

DomoFullAuth(
    domo_instance: str,
    domo_username: str,
    domo_password: str,
    token_name: str | None = None,
    token: str | None = None,
    user_id: str | None = None,
    is_valid_token: bool = False,
    cache_responses: bool = False,
    cache_size: int = 1000,
    cache_default_ttl: int = 300,
    cache_strategy: str | None = None,
    custom_cache_rules: dict[str, list[str]] | None = None,
)

Bases: DomoAuth

Full authentication using Domo username and password.

This class provides authentication using Domo credentials (username and password) to obtain session tokens. It's typically used for direct user authentication where username/password login is permitted.

Attributes:

Name	Type	Description
`domo_instance`	`str`	The Domo instance identifier
`domo_username`	`str`	Domo username for authentication
`domo_password`	`str`	Domo password for authentication (not shown in repr)
`token_name`	`str \| None`	Name identifier for the token
`token`	`str \| None`	The authentication token (not shown in repr)
`user_id`	`str \| None`	The authenticated user's ID
`is_valid_token`	`bool`	Whether the current token is valid
`cache_responses`	`bool`	Enable HTTP response caching (default: False)
`cache_size`	`int`	Maximum number of cache entries (default: 1000)
`cache_default_ttl`	`int`	Default cache TTL in seconds (default: 300)
`cache_strategy`	`str \| None`	Cache invalidation strategy (default: None, uses SMART)
`custom_cache_rules`	`dict \| None`	Custom cache invalidation rules (default: None)

Example

auth = DomoFullAuth( ... domo_instance="mycompany", ... domo_username="user@company.com", ... domo_password="", # pragma: allowlist secret ... ) token = await auth.get_auth_token()

auth_header `property` ¶

auth_header: dict

Generate the full authentication header specific to product APIs.

get_auth_token `async` ¶

get_auth_token(
    return_raw: bool = False, **context_kwargs
) -> str | ResponseGetData

Retrieve the authentication token from product APIs using the provided credentials.

Source code in src/crew_dcs/auth/full.py

@log_call(
    level_name="auth",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_auth_token(
    self,
    return_raw: bool = False,
    **context_kwargs,
) -> str | ResponseGetData:
    """Retrieve the authentication token from product APIs using the provided credentials."""
    from ..routes import auth as auth_routes

    res = await auth_routes.get_full_auth(
        auth=None,
        domo_instance=self.domo_instance,
        domo_username=self.domo_username,
        domo_password=self.domo_password,
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )

    if return_raw:
        return res

    if res.is_success and res.response:
        self.is_valid_token = True

        token = str(res.response.get("sessionToken", ""))
        self.token = token
        self.token_name = self.token_name or "full_auth"

    if not self.token:
        raise AuthError(message="Failed to retrieve authentication token")

    return self.token

DomoJupyterAuth ¶

Mixin for Jupyter auth. Not instantiable alone.

Provides jupyter_token, service_location, service_prefix validation and jupyter_auth_header. Use DomoJupyterFullAuth or DomoJupyterTokenAuth.

jupyter_auth_header `property` ¶

jupyter_auth_header: dict

Header for Jupyter server API (Authorization: Token ...).

DomoJupyterFullAuth `dataclass` ¶

DomoJupyterFullAuth(
    jupyter_token: str,
    service_location: str,
    service_prefix: str,
    domo_instance: str,
    domo_username: str,
    domo_password: str,
    token_name: str | None = None,
    token: str | None = None,
    user_id: str | None = None,
    is_valid_token: bool = False,
    cache_responses: bool = False,
    cache_size: int = 1000,
    cache_default_ttl: int = 300,
    cache_strategy: str | None = None,
    custom_cache_rules: dict[str, list[str]] | None = None,
)

Bases: DomoJupyterAuth, DomoAuth

Jupyter authentication using full credentials (username/password).

auth_header `property` ¶

auth_header: dict

Combined Domo + Jupyter headers (no guard — Jupyter token always present).

convert_auth `classmethod` ¶

convert_auth(
    auth: DomoFullAuth,
    jupyter_token: str,
    service_location: str,
    service_prefix: str,
) -> DomoJupyterFullAuth

Create DomoJupyterFullAuth from DomoFullAuth + Jupyter params.

Source code in src/crew_dcs/auth/jupyter.py

@classmethod
def convert_auth(
    cls,
    auth: DomoFullAuth,
    jupyter_token: str,
    service_location: str,
    service_prefix: str,
) -> "DomoJupyterFullAuth":
    """Create DomoJupyterFullAuth from DomoFullAuth + Jupyter params."""
    return cls(
        domo_instance=auth.domo_instance,
        domo_username=auth.domo_username,
        domo_password=auth.domo_password,
        jupyter_token=jupyter_token,
        service_location=service_location,
        service_prefix=service_prefix,
        token_name=auth.token_name,
        token=auth.token,
        user_id=auth.user_id,
        is_valid_token=auth.is_valid_token,
    )

get_auth_token `async` ¶

get_auth_token(
    return_raw: bool = False, **context_kwargs
) -> str | ResponseGetData

Retrieve the authentication token from product APIs.

Source code in src/crew_dcs/auth/jupyter.py

@log_call(
    level_name="auth",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_auth_token(
    self,
    return_raw: bool = False,
    **context_kwargs,
) -> str | ResponseGetData:
    """Retrieve the authentication token from product APIs."""
    from ..routes import auth as auth_routes

    res = await auth_routes.get_full_auth(
        auth=None,
        domo_instance=self.domo_instance,
        domo_username=self.domo_username,
        domo_password=self.domo_password,
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )

    if return_raw:
        return res

    if res.is_success and res.response:
        self.is_valid_token = True
        token = str(res.response.get("sessionToken", ""))
        self.token = token
        self.token_name = self.token_name or "full_auth"

    if not self.token:
        raise AuthError(message="Failed to retrieve authentication token")

    return self.token

DomoJupyterTokenAuth `dataclass` ¶

DomoJupyterTokenAuth(
    jupyter_token: str,
    service_location: str,
    service_prefix: str,
    domo_instance: str,
    domo_access_token: str,
    token_name: str | None = None,
    token: str | None = None,
    user_id: str | None = None,
    is_valid_token: bool = False,
    cache_responses: bool = False,
    cache_size: int = 1000,
    cache_default_ttl: int = 300,
    cache_strategy: str | None = None,
    custom_cache_rules: dict[str, list[str]] | None = None,
)

Bases: DomoJupyterAuth, DomoAuth

Jupyter authentication using access tokens.

auth_header `property` ¶

auth_header: dict

Combined Domo + Jupyter headers (no guard — both tokens always present).

convert_auth `classmethod` ¶

convert_auth(
    auth: DomoTokenAuth,
    jupyter_token: str,
    service_location: str,
    service_prefix: str,
) -> DomoJupyterTokenAuth

Create DomoJupyterTokenAuth from DomoTokenAuth + Jupyter params.

Source code in src/crew_dcs/auth/jupyter.py

@classmethod
def convert_auth(
    cls,
    auth: DomoTokenAuth,
    jupyter_token: str,
    service_location: str,
    service_prefix: str,
) -> "DomoJupyterTokenAuth":
    """Create DomoJupyterTokenAuth from DomoTokenAuth + Jupyter params."""
    return cls(
        domo_instance=auth.domo_instance,
        domo_access_token=auth.domo_access_token,
        jupyter_token=jupyter_token,
        service_location=service_location,
        service_prefix=service_prefix,
        token_name=auth.token_name,
        token=auth.token,
        user_id=auth.user_id,
        is_valid_token=auth.is_valid_token,
    )

get_auth_token `async` ¶

get_auth_token(
    token_name: str | None = None, **context_kwargs
) -> str

Retrieve the access token, validating via who_am_i if needed.

Source code in src/crew_dcs/auth/jupyter.py

@log_call(
    level_name="auth",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_auth_token(
    self,
    token_name: str | None = None,
    **context_kwargs,
) -> str:
    """Retrieve the access token, validating via who_am_i if needed."""
    if not self.user_id:
        await self.who_am_i(**context_kwargs)
    self.token = self.domo_access_token
    self.is_valid_token = True
    if token_name:
        self.token_name = token_name
    return self.token

DomoTokenAuth `dataclass` ¶

DomoTokenAuth(
    domo_access_token: str,
    domo_instance: str,
    token_name: str | None = None,
    token: str | None = None,
    user_id: str | None = None,
    is_valid_token: bool = False,
    cache_responses: bool = False,
    cache_size: int = 1000,
    cache_default_ttl: int = 300,
    cache_strategy: str | None = None,
    custom_cache_rules: dict[str, list[str]] | None = None,
)

Bases: DomoAuth

Token-based authentication using pre-generated access tokens.

This authentication method uses access tokens generated from Domo's admin panel (Admin > Access Tokens). This is particularly useful in environments where direct username/password authentication is not permitted or for automated systems.

Attributes:

Name	Type	Description
`domo_access_token`	`str`	Pre-generated access token (not shown in repr)
`domo_instance`	`str`	The Domo instance identifier
`token_name`	`str \| None`	Name identifier for the token
`token`	`str \| None`	The authentication token (not shown in repr)
`user_id`	`str \| None`	The authenticated user's ID
`is_valid_token`	`bool`	Whether the current token is valid
`cache_responses`	`bool`	Enable HTTP response caching (default: False)
`cache_size`	`int`	Maximum number of cache entries (default: 1000)
`cache_default_ttl`	`int`	Default cache TTL in seconds (default: 300)
`cache_strategy`	`str \| None`	Cache invalidation strategy (default: None, uses SMART)
`custom_cache_rules`	`dict \| None`	Custom cache invalidation rules (default: None)

Example

auth = DomoTokenAuth( ... domo_access_token="your-access-token-here", ... domo_instance="mycompany" ... ) token = await auth.get_auth_token()

auth_header `property` ¶

auth_header: dict

Generate the authentication header for access token based authentication.

get_auth_token `async` ¶

get_auth_token(
    token_name: str | None = None, **context_kwargs
) -> str

Retrieve the access token, updating internal attributes as necessary.

Source code in src/crew_dcs/auth/token.py

@log_call(
    level_name="auth",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_auth_token(
    self,
    token_name: str | None = None,
    **context_kwargs,
) -> str:
    """Retrieve the access token, updating internal attributes as necessary."""
    if not self.user_id:
        await self.who_am_i(**context_kwargs)

    self.token = self.domo_access_token
    self.is_valid_token = True

    if token_name:
        self.token_name = token_name

    return self.token

test_is_full_auth ¶

test_is_full_auth(
    auth,
    function_name: str | None = None,
    debug_num_stacks_to_drop: int = 1,
) -> None

Test that the provided object is a DomoFullAuth instance.

This validation function ensures that the authentication object is of the correct type for functions that specifically require full authentication.

Parameters:

Name	Type	Description	Default
`auth`		The authentication object to validate	required
`function_name`	`str \| None`	Override function name for error reporting	`None`
`debug_num_stacks_to_drop`	`int`	Number of stack frames to drop for debugging	`1`

Raises:

Type	Description
`AuthError`	If auth is not a DomoFullAuth instance

Source code in src/crew_dcs/auth/utils.py

@log_call(
    level_name="auth",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
def test_is_full_auth(
    auth,
    function_name: str | None = None,
    debug_num_stacks_to_drop: int = 1,
) -> None:
    """Test that the provided object is a DomoFullAuth instance.

    This validation function ensures that the authentication object is of the
    correct type for functions that specifically require full authentication.

    Args:
        auth: The authentication object to validate
        function_name (str | None): Override function name for error reporting
        debug_num_stacks_to_drop (int): Number of stack frames to drop for debugging

    Raises:
        AuthError: If auth is not a DomoFullAuth instance
    """
    from .full import DomoFullAuth

    function_name = function_name or "test_is_full_auth"
    if not isinstance(auth, DomoFullAuth):
        raise AuthError(
            message=f"{function_name} requires DomoFullAuth authentication."
        )

test_is_jupyter_auth ¶

test_is_jupyter_auth(
    auth, required_auth_type_ls: list | None = None
) -> None

Test that the provided object is a valid Jupyter authentication instance.

This validation function ensures that the authentication object is one of the accepted Jupyter authentication types for functions that specifically require Jupyter authentication capabilities.

Parameters:

Name	Type	Description	Default
`auth`		The authentication object to validate	required
`required_auth_type_ls`	`list \| None`	list of acceptable auth types. Defaults to [DomoJupyterFullAuth, DomoJupyterTokenAuth]	`None`

Raises:

Type	Description
`AuthError`	If auth is not one of the required Jupyter auth types

Source code in src/crew_dcs/auth/utils.py

@log_call(
    level_name="auth",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
def test_is_jupyter_auth(
    auth,
    required_auth_type_ls: list | None = None,
) -> None:
    """Test that the provided object is a valid Jupyter authentication instance.

    This validation function ensures that the authentication object is one of the
    accepted Jupyter authentication types for functions that specifically require
    Jupyter authentication capabilities.

    Args:
        auth: The authentication object to validate
        required_auth_type_ls (list | None): list of acceptable auth types.
            Defaults to [DomoJupyterFullAuth, DomoJupyterTokenAuth]

    Raises:
        AuthError: If auth is not one of the required Jupyter auth types
    """
    from .jupyter import DomoJupyterAuth

    if required_auth_type_ls is None:
        if not isinstance(auth, DomoJupyterAuth):
            raise AuthError(
                message="test_is_jupyter_auth requires DomoJupyterFullAuth or DomoJupyterTokenAuth authentication, "
                f"got {type(auth).__name__}",
                function_name="test_is_jupyter_auth",
                domo_instance=getattr(auth, "domo_instance", None),
            )
        return

    if not any(isinstance(auth, t) for t in required_auth_type_ls):
        raise AuthError(
            message=f"test_is_jupyter_auth requires {[t.__name__ for t in required_auth_type_ls]} authentication, got {type(auth).__name__}",
            function_name="test_is_jupyter_auth",
            domo_instance=getattr(auth, "domo_instance", None),
        )

auth

auth ¶

DomoAuth ¶

auth_header abstractmethod property ¶

url_manual_login property ¶

cache_stats ¶

clear_cache ¶

close_session async ¶

elevate_otp async ¶

get_auth_token abstractmethod async ¶

invalidate_cache async ¶

print_is_token async ¶

set_cache_strategy ¶

who_am_i async ¶

DomoDeveloperAuth dataclass ¶

auth_header property ¶

get_auth_token async ¶

who_am_i async ¶

DomoFullAuth dataclass ¶

auth_header property ¶

get_auth_token async ¶

DomoJupyterAuth ¶

jupyter_auth_header property ¶

DomoJupyterFullAuth dataclass ¶

auth_header property ¶

convert_auth classmethod ¶

get_auth_token async ¶

DomoJupyterTokenAuth dataclass ¶

auth_header property ¶

convert_auth classmethod ¶

get_auth_token async ¶

DomoTokenAuth dataclass ¶

auth_header property ¶

get_auth_token async ¶

test_is_full_auth ¶

test_is_jupyter_auth ¶

Modules¶

auth_header `abstractmethod` `property` ¶

url_manual_login `property` ¶

close_session `async` ¶

elevate_otp `async` ¶

get_auth_token `abstractmethod` `async` ¶

invalidate_cache `async` ¶

print_is_token `async` ¶

who_am_i `async` ¶

DomoDeveloperAuth `dataclass` ¶

auth_header `property` ¶

get_auth_token `async` ¶

who_am_i `async` ¶

DomoFullAuth `dataclass` ¶

auth_header `property` ¶

get_auth_token `async` ¶

jupyter_auth_header `property` ¶

DomoJupyterFullAuth `dataclass` ¶

auth_header `property` ¶

convert_auth `classmethod` ¶

get_auth_token `async` ¶

DomoJupyterTokenAuth `dataclass` ¶

auth_header `property` ¶

convert_auth `classmethod` ¶

get_auth_token `async` ¶

DomoTokenAuth `dataclass` ¶

auth_header `property` ¶

get_auth_token `async` ¶