history

history

DomoDataflow_History dataclass

DomoDataflow_History(
    auth: DomoAuth,
    dataflow_id: str,
    dataflow: Any = None,
    execution_history: (
        list[DomoDataflow_History_Execution] | None
    ) = None,
)

get_execution_history async

get_execution_history(
    auth: DomoAuth | None = None,
    maximum: int = 10,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoDataflow_History_Execution]

retrieves metadata about execution history. includes details like execution status.

Source code in src/crew_dcs/classes/DomoDataflow/history.py

async def get_execution_history(
    self,
    auth: DomoAuth | None = None,
    maximum: int = 10,  # maximum number of execution histories to retrieve
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoDataflow_History_Execution]:
    """retrieves metadata about execution history.
    includes details like execution status.
    """

    auth = auth or self.auth or self.dataflow.auth

    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    res = await dataflow_routes.get_dataflow_execution_history(
        auth=auth,
        dataflow_id=self.dataflow_id,
        maximum=maximum,
        context=context,
    )

    if return_raw:
        return res

    execution_history = [
        DomoDataflow_History_Execution.from_dict(df_obj, auth)
        for df_obj in res.response
    ]

    await ce.gather_with_concurrency(
        *[
            domo_execution.get_actions(context=context)
            for domo_execution in execution_history
        ],
        n=20,
    )

    self.execution_history = execution_history

    return self.execution_history

DomoDataflow_History_Execution dataclass

DomoDataflow_History_Execution(
    auth: DomoAuth,
    id: str,
    dataflow_id: str,
    dataflow_execution_id: str,
    dataflow_version: str,
    is_failed: bool,
    state: str,
    activation_type: str,
    data_processor: str,
    telemetry: dict,
    execution_stats: dict,
    begin_time: datetime | None = None,
    last_updated: datetime | None = None,
    end_time: datetime | None = None,
    action_results: (
        list[DomoDataflow_ActionResult] | None
    ) = None,
)

get_actions async

get_actions(
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoDataflow_ActionResult]

retrieves details execution action results

Source code in src/crew_dcs/classes/DomoDataflow/history.py

async def get_actions(
    self,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoDataflow_ActionResult]:
    """retrieves details execution action results"""

    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    res = await dataflow_routes.get_dataflow_execution_by_id(
        auth=self.auth,
        dataflow_id=self.dataflow_id,
        execution_id=self.id,
        context=context,
    )

    if return_raw:
        return res

    self.action_results = self._process_results(
        res.response.get("actionResults", [])
    )

    return self.action_results

get_by_id async classmethod

get_by_id(
    auth: DomoAuth,
    dataflow_id: int,
    execution_id: int,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoDataflow_History_Execution

retrieves details about a dataflow execution including actions

Source code in src/crew_dcs/classes/DomoDataflow/history.py

@classmethod
async def get_by_id(
    cls,
    auth: DomoAuth,
    dataflow_id: int,
    execution_id: int,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoDataflow_History_Execution:
    """retrieves details about a dataflow execution including actions"""

    base_context = RouteContext.build_context(
        parent_class=cls.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    res = await dataflow_routes.get_dataflow_execution_by_id(
        auth=auth,
        dataflow_id=dataflow_id,
        execution_id=execution_id,
        context=context,
    )

    if return_raw:
        return res

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