DomoApplication

DomoApplication

DomoApplication package - Classes for Domo Applications and Jobs.

Application_CRUD_Error

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

Bases: RouteError

Raised when application create, update, or delete operations fail.

This exception is used for failures during application job creation, modification, or execution operations.

Source code in src/crew_dcs/routes/application.py

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

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

Application_GET_Error

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

Bases: RouteError

Raised when application retrieval operations fail.

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

Source code in src/crew_dcs/routes/application.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 application {entity_id}"
        else:
            message = "Failed to retrieve applications"

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

DomoJob dataclass

DomoJob(
    auth: DomoAuth,
    id: str,
    raw: dict = dict(),
    application_id: str = None,
    name: str = None,
    user_id: str = None,
    logs_dataset_id: str = None,
    execution_timeout: int = 1440,
    is_enabled: bool = False,
    customer_id: str = None,
    created_dt: datetime | None = None,
    updated_dt: datetime | None = None,
    description: str = None,
    execution_payload: dict = dict(),
    share_state: dict = dict(),
    accounts: list[str] = list(),
    triggers: list[DomoTrigger] = list(),
)

Bases: DomoEntity

A class for interacting with Domo Application Jobs.

This class represents a job within a Domo application, providing methods for retrieving, creating, updating, and executing jobs.

Attributes:

Name	Type	Description
auth	DomoAuth	Authentication object for API requests
id	str	Unique identifier for the job (job_id)
application_id	str	ID of the parent application
name	str	Name of the job
raw	dict	Raw API response data
user_id	str	ID of the user who created the job
logs_dataset_id	str	Dataset ID where job logs are stored
execution_timeout	int	Timeout in minutes for job execution
is_enabled	bool	Whether the job is enabled (has triggers)
customer_id	str	Customer ID
created_dt	datetime | None	Job creation datetime
updated_dt	datetime | None	Job last update datetime
description	str	Job description
execution_payload	dict	Job execution configuration
share_state	dict	Job sharing state
accounts	list[str]	list of account IDs associated with the job
triggers	list[DomoTrigger]	list of job triggers

display_url property

display_url: str

Generate the URL to display this job in the Domo interface.

Returns:

Name	Type	Description
str	str	URL to view the job in Domo

create async classmethod

create(
    auth: DomoAuth,
    application_id: str,
    job_name: str,
    execution_payload: dict,
    job_description: str | None = None,
    execution_timeout: int = 1440,
    accounts: list[str] | None = None,
    triggers: list[dict] | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Create a new job in the application.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object	required
application_id	str	ID of the parent application	required
job_name	str	Name for the new job	required
execution_payload	dict	Job execution configuration	required
job_description	str | None	Optional job description	None
execution_timeout	int	Timeout in minutes (default 1440)	1440
accounts	list[str] | None	list of account IDs	None
triggers	list[dict] | None	list of job triggers	None
return_raw	bool	Return raw ResponseGetData instead of DomoJob	False
debug_api	bool	Enable API debugging	False
session	AsyncClient | None	Optional httpx session for request pooling	None
debug_num_stacks_to_drop	int	Number of stack frames to drop in debug output	2

Returns:

Type	Description
	DomoJob or ResponseGetData: Created job object or raw response

Raises:

Type	Description
CRUD_ApplicationJob_Error	If creation fails

Source code in src/crew_dcs/classes/DomoApplication/Job.py

@classmethod
async def create(
    cls,
    auth: DomoAuth,
    application_id: str,
    job_name: str,
    execution_payload: dict,
    job_description: str | None = None,
    execution_timeout: int = 1440,
    accounts: list[str] | None = None,
    triggers: list[dict] | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Create a new job in the application.

    Args:
        auth: Authentication object
        application_id: ID of the parent application
        job_name: Name for the new job
        execution_payload: Job execution configuration
        job_description: Optional job description
        execution_timeout: Timeout in minutes (default 1440)
        accounts: list of account IDs
        triggers: list of job triggers
        return_raw: Return raw ResponseGetData instead of DomoJob
        debug_api: Enable API debugging
        session: Optional httpx session for request pooling
        debug_num_stacks_to_drop: Number of stack frames to drop in debug output

    Returns:
        DomoJob or ResponseGetData: Created job object or raw response

    Raises:
        CRUD_ApplicationJob_Error: If creation fails
    """
    body = {
        "jobName": job_name,
        "jobDescription": job_description,
        "executionTimeout": execution_timeout,
        "executionPayload": execution_payload,
        "accounts": accounts or [],
        "triggers": triggers or [],
    }

    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 application_routes.create_application_job(
        auth=auth,
        body=body,
        application_id=application_id,
        context=context,
    )

    if return_raw:
        return res

    # Get the created job
    job_id = res.response.get("jobId")
    return await cls.get_by_id(
        auth=auth,
        application_id=application_id,
        job_id=job_id,
        context=context,
    )

execute async

execute(
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Execute this job.

Parameters:

Name	Type	Description	Default
return_raw	bool	Return raw ResponseGetData	False
debug_api	bool	Enable API debugging	False
session	AsyncClient | None	Optional httpx session for request pooling	None
debug_num_stacks_to_drop	int	Number of stack frames to drop in debug output	2

Returns:

Name	Type	Description
ResponseGetData		Response from job execution

Raises:

Type	Description
CRUD_ApplicationJob_Error	If execution fails

Source code in src/crew_dcs/classes/DomoApplication/Job.py

async def execute(
    self,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Execute this job.

    Args:
        return_raw: Return raw ResponseGetData
        debug_api: Enable API debugging
        session: Optional httpx session for request pooling
        debug_num_stacks_to_drop: Number of stack frames to drop in debug output

    Returns:
        ResponseGetData: Response from job execution

    Raises:
        CRUD_ApplicationJob_Error: If execution 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 application_routes.execute_application_job(
        auth=self.auth,
        application_id=self.application_id,
        job_id=self.id,
        context=context,
    )

    if return_raw:
        return res

    return res

from_dict classmethod

from_dict(auth: DomoAuth, obj: dict[str, Any])

Create a DomoJob instance from a dictionary representation.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object	required
obj	dict[str, Any]	Dictionary containing job data from API	required

Returns:

Name	Type	Description
DomoJob		New instance with populated attributes

Source code in src/crew_dcs/classes/DomoApplication/Job.py

@classmethod
def from_dict(cls, auth: DomoAuth, obj: dict[str, Any]):
    """Create a DomoJob instance from a dictionary representation.

    Args:
        auth: Authentication object
        obj: Dictionary containing job data from API

    Returns:
        DomoJob: New instance with populated attributes
    """
    # Use DomoJob_Base conversion logic
    base_obj = DomoJob_Base._convert_API_res_to_DomoJob_base_obj(obj=obj)

    return cls(
        auth=auth,
        raw=obj,
        **base_obj,
    )

get_by_id async classmethod

get_by_id(
    auth: DomoAuth,
    application_id: str,
    job_id: str,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Retrieve a job by its ID.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object	required
application_id	str	ID of the parent application	required
job_id	str	Unique identifier of the job	required
return_raw	bool	Return raw ResponseGetData instead of DomoJob	False
debug_api	bool	Enable API debugging	False
session	AsyncClient | None	Optional httpx session for request pooling	None
debug_num_stacks_to_drop	int	Number of stack frames to drop in debug output	2

Returns:

Type	Description
	DomoJob or ResponseGetData: The job object or raw response

Raises:

Type	Description
ApplicationError_NoneRetrieved	If job retrieval fails

Source code in src/crew_dcs/classes/DomoApplication/Job.py

@classmethod
async def get_by_id(
    cls,
    auth: DomoAuth,
    application_id: str,
    job_id: str,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Retrieve a job by its ID.

    Args:
        auth: Authentication object
        application_id: ID of the parent application
        job_id: Unique identifier of the job
        return_raw: Return raw ResponseGetData instead of DomoJob
        debug_api: Enable API debugging
        session: Optional httpx session for request pooling
        debug_num_stacks_to_drop: Number of stack frames to drop in debug output

    Returns:
        DomoJob or ResponseGetData: The job object or raw response

    Raises:
        ApplicationError_NoneRetrieved: If job 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 application_routes.get_application_job_by_id(
        auth=auth,
        application_id=application_id,
        job_id=job_id,
        context=context,
    )

    if return_raw:
        return res

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

get_entity_by_id async classmethod

get_entity_by_id(
    auth: DomoAuth,
    entity_id: str,
    debug_num_stacks_to_drop: int = 2,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    **kwargs
)

Fetch a job by its entity ID.

Note: For DomoJob, this requires both application_id and job_id. The entity_id is treated as the job_id, and application_id must be provided in kwargs.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object	required
entity_id	str	Job ID (job_id)	required
debug_num_stacks_to_drop	int	Number of stack frames to drop in debug	2
debug_api	bool	Enable API debugging	False
session	AsyncClient | None	Optional httpx session	None
**kwargs		Must include application_id	{}

Returns:

Type	Description
	DomoJob instance

Raises:

Type	Description
ValueError	If application_id not provided in kwargs

Source code in src/crew_dcs/classes/DomoApplication/Job.py

@classmethod
async def get_entity_by_id(
    cls,
    auth: DomoAuth,
    entity_id: str,
    debug_num_stacks_to_drop: int = 2,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    **kwargs,
):
    """Fetch a job by its entity ID.

    Note: For DomoJob, this requires both application_id and job_id.
    The entity_id is treated as the job_id, and application_id must be
    provided in kwargs.

    Args:
        auth: Authentication object
        entity_id: Job ID (job_id)
        debug_num_stacks_to_drop: Number of stack frames to drop in debug
        debug_api: Enable API debugging
        session: Optional httpx session
        **kwargs: Must include application_id

    Returns:
        DomoJob instance

    Raises:
        ValueError: If application_id not provided in kwargs
    """
    application_id = kwargs.get("application_id")
    if not application_id:
        raise ValueError(
            "application_id is required in kwargs for DomoJob.get_entity_by_id()"
        )

    return await cls.get_by_id(
        auth=auth,
        application_id=application_id,
        job_id=entity_id,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        debug_api=debug_api,
        session=session,
        **kwargs,
    )

to_dict

to_dict() -> dict

Convert the job to a dictionary for API requests.

Returns:

Name	Type	Description
dict	dict	Dictionary representation of the job

Source code in src/crew_dcs/classes/DomoApplication/Job.py

def to_dict(self) -> dict:
    """Convert the job to a dictionary for API requests.

    Returns:
        dict: Dictionary representation of the job
    """
    trigger_ls = (
        [self.triggers[0].schedule.to_dict()] if len(self.triggers) > 0 else []
    )

    execution_payload = self.execution_payload or {}
    if self.logs_dataset_id:
        execution_payload.update({"metricsDatasetId": self.logs_dataset_id})

    return {
        "jobId": self.id,
        "jobName": self.name,
        "userId": self.user_id,
        "applicationId": self.application_id,
        "customerId": self.customer_id,
        "executionTimeout": self.execution_timeout,
        "executionPayload": execution_payload,
        "shareState": self.share_state,
        "triggers": trigger_ls,
        "jobDescription": self.description,
        "accounts": self.accounts,
    }

update async

update(
    body: dict | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Update this job's configuration.

Parameters:

Name	Type	Description	Default
body	dict | None	Dictionary containing job update data (uses to_dict if None)	None
return_raw	bool	Return raw ResponseGetData instead of DomoJob	False
debug_api	bool	Enable API debugging	False
session	AsyncClient | None	Optional httpx session for request pooling	None
debug_num_stacks_to_drop	int	Number of stack frames to drop in debug output	2

Returns:

Type	Description
	DomoJob or ResponseGetData: Updated job object or raw response

Raises:

Type	Description
CRUD_ApplicationJob_Error	If update fails

Source code in src/crew_dcs/classes/DomoApplication/Job.py

async def update(
    self,
    body: dict | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Update this job's configuration.

    Args:
        body: Dictionary containing job update data (uses to_dict if None)
        return_raw: Return raw ResponseGetData instead of DomoJob
        debug_api: Enable API debugging
        session: Optional httpx session for request pooling
        debug_num_stacks_to_drop: Number of stack frames to drop in debug output

    Returns:
        DomoJob or ResponseGetData: Updated job object or raw response

    Raises:
        CRUD_ApplicationJob_Error: If update fails
    """
    if body is None:
        body = self.to_dict()

    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 application_routes.update_application_job(
        auth=self.auth,
        body=body,
        job_id=self.id,
        application_id=self.application_id,
        context=context,
    )

    if return_raw:
        return res

    # Refresh the job data
    return await self.get_by_id(
        auth=self.auth,
        application_id=self.application_id,
        job_id=self.id,
        context=context,
    )

DomoJob_Base dataclass

DomoJob_Base(
    auth: DomoAuth,
    id: str = None,
    raw: dict = dict(),
    name: str = None,
    application_id: str = None,
    logs_dataset_id: str = None,
    user_id: str = None,
    execution_timeout: int = 1440,
    is_enabled: bool = False,
    customer_id: str = None,
    created_dt: datetime = None,
    updated_dt: datetime = None,
    description: str = None,
    execution_payload: dict = dict(),
    share_state: dict = dict(),
    accounts: list[str] = list(),
    triggers: list[DomoTrigger] = list(),
)

Bases: DomoEntity

the base class only captures attributes applicable to all jobs (i.e. does not destructure execution_payload onto the class) build Application / Job extensions by creating children of the DomoJob_Base class

display_url property

display_url: str

Generate the URL to display this job in the Domo interface.

Returns:

Name	Type	Description
str	str	Complete URL to view the job in Domo

from_dict classmethod

from_dict(auth: DomoAuth, obj: dict[str, Any])

Create a DomoJob_Base instance from an API response dictionary.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for API requests	required
obj	dict[str, Any]	Dictionary representation of the job from API response	required

Returns:

Type	Description
	DomoJob_Base instance

Source code in src/crew_dcs/classes/DomoApplication/Job_Base.py

@classmethod
def from_dict(
    cls,
    auth: DomoAuth,
    obj: dict[str, Any],
):
    """Create a DomoJob_Base instance from an API response dictionary.

    Args:
        auth: Authentication object for API requests
        obj: Dictionary representation of the job from API response

    Returns:
        DomoJob_Base instance
    """

    base_obj = cls._convert_API_res_to_DomoJob_base_obj(obj=obj)

    return cls(
        auth=auth,
        raw=obj,
        **base_obj,
    )

get_by_id async classmethod

get_by_id(
    auth: DomoAuth,
    application_id: str,
    job_id: str,
    session: AsyncClient | None = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Retrieve a job by its ID from a specific application.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for API requests	required
application_id	str	The ID of the application containing the job	required
job_id	str	The ID of the job to retrieve	required
session	AsyncClient | None	Optional HTTP session for connection pooling	None
debug_api	bool	Enable API debugging output	False
debug_num_stacks_to_drop	int	Number of stack frames to drop in debugging	2
return_raw	bool	Return raw response without processing	False

Returns:

Type	Description
	DomoJob_Base instance or ResponseGetData if return_raw=True

Raises:

Type	Description
ApplicationNoJobRetrievedError	If job retrieval fails

Source code in src/crew_dcs/classes/DomoApplication/Job_Base.py

@classmethod
async def get_by_id(
    cls,
    auth: DomoAuth,
    application_id: str,
    job_id: str,
    session: httpx.AsyncClient | None = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """
    Retrieve a job by its ID from a specific application.

    Args:
        auth: Authentication object for API requests
        application_id: The ID of the application containing the job
        job_id: The ID of the job to retrieve
        session: Optional HTTP session for connection pooling
        debug_api: Enable API debugging output
        debug_num_stacks_to_drop: Number of stack frames to drop in debugging
        return_raw: Return raw response without processing

    Returns:
        DomoJob_Base instance or ResponseGetData if return_raw=True

    Raises:
        ApplicationNoJobRetrievedError: If job retrieval fails
    """

    return await cls._get_by_id(
        auth=auth,
        application_id=application_id,
        job_id=job_id,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        return_raw=return_raw,
        new_cls=cls,
        context=context,
        **context_kwargs,
    )

to_dict

to_dict()

this is an abstract method, each DomoJob_Base implementation must define a to_dict() function

Source code in src/crew_dcs/classes/DomoApplication/Job_Base.py

def to_dict(self):
    """this is an abstract method, each DomoJob_Base implementation must define a to_dict() function"""
    return self._generate_to_dict()

Modules

• Application
• Job
• Job_Base
• Job_RemoteDomoStats
• Job_Watchdog