Job_Base

Job_Base

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,
    )

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()