Skip to content

exceptions

JobExecutionError

Bases: RuntimeError

JobExecutionError

Attributes:

Name Type Description
info str | None

A free field for additional information
Source code in fractal_server/app/runner/exceptions.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
class JobExecutionError(RuntimeError):
    """
    JobExecutionError

    Attributes:
        info:
            A free field for additional information
    """

    info: str | None = None

    def __init__(
        self,
        *args,
        info: str | None = None,
    ):
        super().__init__(*args)
        self.info = info

    def assemble_error(self) -> str:
        if self.info:
            content = f"\n{self.info}\n\n"
        else:
            content = str(self)
        message = f"JobExecutionError\n{content}"
        return message

TaskExecutionError

Bases: RuntimeError

Forwards errors occurred during the execution of a task

This error wraps and forwards errors occurred during the execution of tasks, when the exit code is larger than 0 (i.e. the error took place within the task). This error also adds information that is useful to track down and debug the failing task within a workflow.

Attributes:

Name Type Description
workflow_task_id int | None

ID of the workflow task that failed.
workflow_task_order int | None

Order of the task within the workflow.
task_name str | None

Human readable name of the failing task.
Source code in fractal_server/app/runner/exceptions.py 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class TaskExecutionError(RuntimeError):
    """
    Forwards errors occurred during the execution of a task

    This error wraps and forwards errors occurred during the execution of
    tasks, when the exit code is larger than 0 (i.e. the error took place
    within the task). This error also adds information that is useful to track
    down and debug the failing task within a workflow.

    Attributes:
        workflow_task_id:
            ID of the workflow task that failed.
        workflow_task_order:
            Order of the task within the workflow.
        task_name:
            Human readable name of the failing task.
    """

    workflow_task_id: int | None = None
    workflow_task_order: int | None = None
    task_name: str | None = None

    def __init__(
        self,
        *args,
        workflow_task_id: int | None = None,
        workflow_task_order: int | None = None,
        task_name: str | None = None,
    ):
        super().__init__(*args)
        self.workflow_task_id = workflow_task_id
        self.workflow_task_order = workflow_task_order
        self.task_name = task_name