base

base

Base authentication classes for Domo authentication.

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