api_client

api_client

ApiClient dataclass

ApiClient(
    auth: DomoAuth,
    id: str,
    raw: dict,
    name: str = "",
    client_id: str = "",
    client_secret: str = "",
    owner: DomoUser | None = None,
    scopes: list[ApiClient_ScopeEnum] = list(),
    description: str | None = None,
)

Bases: DomoEntity

get_by_id async classmethod

get_by_id(
    auth: DomoAuth,
    id: str,
    session: AsyncClient | None = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    parent_class: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ApiClient

Retrieve a specific API client by its ID.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing instance and credentials	required
id	str	Unique identifier for the API client	required
session	AsyncClient | None	Optional HTTP client session for connection reuse	None
debug_api	bool	Enable detailed API request/response logging	False
debug_num_stacks_to_drop	int	Number of stack frames to omit in debug output	2
parent_class	str | None	Name of calling class for debugging context	None
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	Optional RouteContext for API call configuration	None

Returns:

Type	Description
ApiClient	ResponseGetData object or ApiClient instance

Raises:

Type	Description
ApiClient_GET_Error	If API client retrieval fails

Source code in src/crew_dcs/classes/DomoInstanceConfig/api_client.py

@classmethod
async def get_by_id(
    cls,
    auth: DomoAuth,
    id: str,
    session: httpx.AsyncClient | None = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    parent_class: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> "ApiClient":
    """
    Retrieve a specific API client by its ID.

    Args:
        auth: Authentication object containing instance and credentials
        id: Unique identifier for the API client
        session: Optional HTTP client session for connection reuse
        debug_api: Enable detailed API request/response logging
        debug_num_stacks_to_drop: Number of stack frames to omit in debug output
        parent_class: Name of calling class for debugging context
        return_raw: Return raw API response without processing
        context: Optional RouteContext for API call configuration

    Returns:
        ResponseGetData object or ApiClient instance

    Raises:
        ApiClient_GET_Error: If API client retrieval fails
    """

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        **context_kwargs,
    )

    res = await get_client_by_id(
        auth=auth,
        client_id=id,
        return_raw=return_raw,
        context=context,
    )

    if return_raw:
        return res

    obj = res.response

    owner = await cls.get_user_by_id(
        user_id=obj["userId"], auth=auth, context=context
    )
    obj["owner"] = owner

    return cls.from_dict(auth=auth, obj=obj)

revoke async

revoke(
    session: AsyncClient | None = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    parent_class: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Revoke (delete) this API client.

Parameters:

Name	Type	Description	Default
session	AsyncClient | None	Optional HTTP client session for connection reuse	None
debug_api	bool	Enable detailed API request/response logging	False
debug_num_stacks_to_drop	int	Number of stack frames to omit in debug output	2
parent_class	str | None	Name of calling class for debugging context	None
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	Optional RouteContext for API call configuration	None

Returns:

Type	Description
ResponseGetData	ResponseGetData object with confirmation message

Raises:

Type	Description
ApiClient_RevokeError	If API client revocation fails

Source code in src/crew_dcs/classes/DomoInstanceConfig/api_client.py

async def revoke(
    self,
    session: httpx.AsyncClient | None = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    parent_class: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Revoke (delete) this API client.

    Args:
        session: Optional HTTP client session for connection reuse
        debug_api: Enable detailed API request/response logging
        debug_num_stacks_to_drop: Number of stack frames to omit in debug output
        parent_class: Name of calling class for debugging context
        return_raw: Return raw API response without processing
        context: Optional RouteContext for API call configuration

    Returns:
        ResponseGetData object with confirmation message

    Raises:
        ApiClient_RevokeError: If API client revocation fails
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        **context_kwargs,
    )

    return await revoke_api_client(
        auth=self.auth,
        client_id=str(self.id),
        return_raw=return_raw,
        context=context,
    )

ApiClient_CRUD_Error

ApiClient_CRUD_Error(
    operation: str = "operation",
    entity_id: str | None = None,
    res: ResponseGetData | None = None,
    message: str | None = None,
    **kwargs
)

Bases: RouteError

Raised when API client create, update, or delete operations fail.

This exception is used for failures during API client creation, modification, or deletion operations.

Source code in src/crew_dcs/routes/instance_config/exceptions.py

def __init__(
    self,
    operation: str = "operation",
    entity_id: str | None = None,
    res: rgd.ResponseGetData | None = None,
    message: str | None = None,
    **kwargs,
):
    if not message:
        if entity_id:
            message = f"API client {operation} failed for client {entity_id}"
        else:
            message = f"API client {operation} operation failed"

    super().__init__(
        message=message,
        entity_id=entity_id,
        res=res,
        additional_context={"operation": operation},
        **kwargs,
    )

ApiClient_GET_Error

ApiClient_GET_Error(
    entity_id: str | None = None,
    res: ResponseGetData | None = None,
    message: str | None = None,
    **kwargs
)

Bases: RouteError

Raised when API client retrieval operations fail.

This exception is used for failures during GET operations on API clients, including API errors and unexpected response formats.

Source code in src/crew_dcs/routes/instance_config/exceptions.py

def __init__(
    self,
    entity_id: str | None = None,
    res: rgd.ResponseGetData | None = None,
    message: str | None = None,
    **kwargs,
):
    if not message:
        if entity_id:
            message = f"Failed to retrieve API client {entity_id}"
        else:
            message = "Failed to retrieve API clients"

    super().__init__(message=message, entity_id=entity_id, res=res, **kwargs)

ApiClients dataclass

ApiClients(
    auth: DomoAuth,
    domo_clients: list[ApiClient] = list(),
    invalid_clients: list[ApiClient] = list(),
    parent: Any = None,
)

Bases: DomoManager

create_for_authorized_user async

create_for_authorized_user(
    client_name: str,
    client_description: str = f"created via DL {today()}",
    scope: list[ApiClient_ScopeEnum] | None = None,
    session: AsyncClient | None = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    parent_class: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> ApiClient

Create a new API client for the authenticated user.

Parameters:

Name	Type	Description	Default
client_name	str	Name for the new API client	required
client_description	str	Optional description for the API client	f'created via DL {today()}'
scope	list[ApiClient_ScopeEnum] | None	list of ApiClient_ScopeEnum values, defaults to [data, audit]	None
session	AsyncClient | None	Optional HTTP client session for connection reuse	None
debug_api	bool	Enable detailed API request/response logging	False
debug_num_stacks_to_drop	int	Number of stack frames to omit in debug output	2
parent_class	str | None	Name of calling class for debugging context	None
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	Optional RouteContext for API call configuration	None

Returns:

Type	Description
ApiClient	ResponseGetData object or ApiClient instance with credentials

Raises:

Type	Description
ApiClient_CRUD_Error	If API client creation fails
SearchApiClient_NotFound	If created client cannot be retrieved

Source code in src/crew_dcs/classes/DomoInstanceConfig/api_client.py

async def create_for_authorized_user(
    self,
    client_name: str,
    client_description: str = f"created via DL {dt.date.today()}",
    scope: list[ApiClient_ScopeEnum] | None = None,
    session: httpx.AsyncClient | None = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    parent_class: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> ApiClient:
    """
    Create a new API client for the authenticated user.

    Args:
        client_name: Name for the new API client
        client_description: Optional description for the API client
        scope: list of ApiClient_ScopeEnum values, defaults to [data, audit]
        session: Optional HTTP client session for connection reuse
        debug_api: Enable detailed API request/response logging
        debug_num_stacks_to_drop: Number of stack frames to omit in debug output
        parent_class: Name of calling class for debugging context
        return_raw: Return raw API response without processing
        context: Optional RouteContext for API call configuration

    Returns:
        ResponseGetData object or ApiClient instance with credentials

    Raises:
        ApiClient_CRUD_Error: If API client creation fails
        SearchApiClient_NotFound: If created client cannot be retrieved
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        **context_kwargs,
    )

    res = await create_api_client(
        auth=self.auth,
        client_name=client_name,
        client_description=client_description,
        scope=scope,
        return_raw=return_raw,
        context=context,
    )

    if return_raw:
        return res

    domo_client = await self.get_by_name(
        client_name=client_name,
        context=context,
    )
    domo_client.client_id = res.response["client_id"]
    domo_client.client_secret = res.response["client_secret"]

    return domo_client

get async

get(
    session: AsyncClient | None = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    parent_class: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[ApiClient] | ResponseGetData

Retrieve all API clients for the authenticated instance.

Parameters:

Name	Type	Description	Default
session	AsyncClient | None	Optional HTTP client session for connection reuse	None
debug_api	bool	Enable detailed API request/response logging	False
debug_num_stacks_to_drop	int	Number of stack frames to omit in debug output	2
parent_class	str | None	Name of calling class for debugging context	None
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	Optional RouteContext for API call configuration	None

Returns:

Type	Description
list[ApiClient] | ResponseGetData	ResponseGetData object or list of ApiClient instances

Raises:

Type	Description
ApiClient_GET_Error	If API client retrieval fails

Source code in src/crew_dcs/classes/DomoInstanceConfig/api_client.py

async def get(
    self,
    session: httpx.AsyncClient | None = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    parent_class: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[ApiClient] | rgd.ResponseGetData:
    """
    Retrieve all API clients for the authenticated instance.

    Args:
        session: Optional HTTP client session for connection reuse
        debug_api: Enable detailed API request/response logging
        debug_num_stacks_to_drop: Number of stack frames to omit in debug output
        parent_class: Name of calling class for debugging context
        return_raw: Return raw API response without processing
        context: Optional RouteContext for API call configuration

    Returns:
        ResponseGetData object or list of ApiClient instances

    Raises:
        ApiClient_GET_Error: If API client retrieval fails
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        **context_kwargs,
    )

    res = await get_api_clients(
        auth=self.auth,
        return_raw=return_raw,
        context=context,
    )

    if return_raw:
        return res

    self.domo_clients = await dmce.gather_with_concurrency(
        *[
            ApiClient.get_by_id(auth=self.auth, id=obj["id"], context=context)
            for obj in res.response
        ],
        n=10,
    )

    if self.parent:
        self.domo_clients = [
            client for client in self.domo_clients if client.owner == self.parent
        ]

    self.invalid_clients = [
        client for client in self.domo_clients if not client.is_valid
    ]

    return self.domo_clients

get_by_name async

get_by_name(
    client_name: str,
    session: AsyncClient | None = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    parent_class: str | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ApiClient

Retrieve an API client by its name.

Parameters:

Name	Type	Description	Default
client_name	str	Name of the API client to find	required
session	AsyncClient | None	Optional HTTP client session for connection reuse	None
debug_api	bool	Enable detailed API request/response logging	False
debug_num_stacks_to_drop	int	Number of stack frames to omit in debug output	2
parent_class	str | None	Name of calling class for debugging context	None
context	RouteContext | None	Optional RouteContext for API call configuration	None

Returns:

Type	Description
ApiClient	ApiClient instance matching the specified name

Raises:

Type	Description
SearchApiClient_NotFound	If no client with the specified name is found
ApiClient_GET_Error	If API client retrieval fails

Source code in src/crew_dcs/classes/DomoInstanceConfig/api_client.py

async def get_by_name(
    self,
    client_name: str,
    session: httpx.AsyncClient | None = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    parent_class: str | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> ApiClient:
    """
    Retrieve an API client by its name.

    Args:
        client_name: Name of the API client to find
        session: Optional HTTP client session for connection reuse
        debug_api: Enable detailed API request/response logging
        debug_num_stacks_to_drop: Number of stack frames to omit in debug output
        parent_class: Name of calling class for debugging context
        context: Optional RouteContext for API call configuration

    Returns:
        ApiClient instance matching the specified name

    Raises:
        SearchApiClient_NotFound: If no client with the specified name is found
        ApiClient_GET_Error: If API client retrieval fails
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop + 1,
        **context_kwargs,
    )

    await self.get(
        context=context,
    )

    domo_client = next(
        (
            _domo_client
            for _domo_client in self.domo_clients
            if _domo_client.name == client_name
        ),
        None,
    )

    if not domo_client:
        raise SearchApiClientNotFoundError(
            search_criteria=f"client name: {client_name}"
        )

    return domo_client

upsert_client async

upsert_client(
    client_name: str,
    client_description: str | None = None,
    scope: list[ApiClient_ScopeEnum] | None = None,
    is_regenerate: bool = False,
    session: AsyncClient | None = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    parent_class: str | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ApiClient

Create or update an API client (upsert operation).

Parameters:

Name	Type	Description	Default
client_name	str	Name of the API client to create or update	required
client_description	str | None	Optional description for the API client	None
scope	list[ApiClient_ScopeEnum] | None	list of ApiClient_ScopeEnum values, defaults to [data, audit]	None
is_regenerate	bool	If True, revoke existing client and create new one	False
session	AsyncClient | None	Optional HTTP client session for connection reuse	None
debug_api	bool	Enable detailed API request/response logging	False
debug_num_stacks_to_drop	int	Number of stack frames to omit in debug output	2
parent_class	str | None	Name of calling class for debugging context	None
context	RouteContext | None	Optional RouteContext for API call configuration	None

Returns:

Type	Description
ApiClient	ApiClient instance (existing or newly created)

Raises:

Type	Description
ApiClient_CRUD_Error	If API client creation or revocation fails

Source code in src/crew_dcs/classes/DomoInstanceConfig/api_client.py

async def upsert_client(
    self,
    client_name: str,
    client_description: str | None = None,
    scope: list[ApiClient_ScopeEnum] | None = None,
    is_regenerate: bool = False,
    session: httpx.AsyncClient | None = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    parent_class: str | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> ApiClient:
    """
    Create or update an API client (upsert operation).

    Args:
        client_name: Name of the API client to create or update
        client_description: Optional description for the API client
        scope: list of ApiClient_ScopeEnum values, defaults to [data, audit]
        is_regenerate: If True, revoke existing client and create new one
        session: Optional HTTP client session for connection reuse
        debug_api: Enable detailed API request/response logging
        debug_num_stacks_to_drop: Number of stack frames to omit in debug output
        parent_class: Name of calling class for debugging context
        context: Optional RouteContext for API call configuration

    Returns:
        ApiClient instance (existing or newly created)

    Raises:
        ApiClient_CRUD_Error: If API client creation or revocation fails
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        **context_kwargs,
    )

    domo_client = None

    with contextlib.suppress(SearchApiClientNotFoundError):
        domo_client = await self.get_by_name(
            client_name=client_name,
            context=context,
        )

    if domo_client:
        if not is_regenerate:
            await logger.info(f"Using existing API client: {client_name}")
            return domo_client

        await logger.info(f"Revoking and recreating API client: {client_name}")
        await domo_client.revoke(
            context=context,
        )
    else:
        await logger.info(f"Creating new API client: {client_name}")

    return await self.create_for_authorized_user(
        client_name=client_name,
        client_description=client_description or "",
        scope=scope,
        context=context,
    )

SearchApiClientNotFoundError

SearchApiClientNotFoundError(
    search_criteria: str,
    res: ResponseGetData | None = None,
    message: str | None = None,
    **kwargs
)

Bases: RouteError

Raised when API client search operations return no results.

This exception is used when searching for API clients by name or other criteria yields no matching results.

Source code in src/crew_dcs/routes/instance_config/exceptions.py

def __init__(
    self,
    search_criteria: str,
    res: rgd.ResponseGetData | None = None,
    message: str | None = None,
    **kwargs,
):
    if not message:
        message = f"No API client found matching: {search_criteria}"

    super().__init__(
        message=message,
        res=res,
        additional_context={"search_criteria": search_criteria},
        **kwargs,
    )