qhana_plugin_runner.db.models.tasks module

class qhana_plugin_runner.db.models.tasks.ProcessingTask(task_name, started_at=datetime.datetime(2026, 6, 3, 8, 11, 23, 779002), finished_at=None, parameters='', data=<factory>, multi_step=False, steps=<factory>, current_step=-1, progress_value=None, progress_start=0, progress_target=100, progress_unit='%', task_status=None, task_log='', outputs=<factory>, links=<factory>, subscriptions=<factory>)

Bases: object

Dataclass for persisting (logical) task information.

id

automatically generated database id. Use the id to fetch this information from the database.

Type:

int, optional

task_name

the name of the (logical) task corresponding to this information object

Type:

str

started_at

the moment the task was scheduled. (default utcnow())

Type:

datetime, optional

finished_at

the moment the task finished successfully or with an error.

Type:

Optional[datetime], optional

parameters

the parameters for the task. Task parameters should already be prepared and error checked before starting the task.

Type:

str

data

mutable JSON-like store for additional lightweight task data. Default value is empty dict.

Type:

JSON_LIKE

steps

ordered list of steps of type Step. Index number automatically increases when new elements are appended. Note: only use add_next_step() to add a new step. Steps must not be deleted.

Type:

OrderingList[Step]

current_step

index of last added step.

Type:

int

progress_value

current progress value. None by default.

Type:

float

progress_start

progress start value.

Type:

float

progress_target

progress target value.

Type:

float

progress_unit

progress unit (default: “%”).

Type:

str

task_status

the status string of the plugin execution, can only be PENDING, SUCCESS, or ERROR.

Type:

Optional[str], optional

task_log

the task log, task metadata or the error of the finished task. All data results should be file outputs of the task!

Type:

str

outputs

the output data (files) of the task

Type:

List[TaskFile], optional

add_next_step(href: str, ui_href: str, step_id: str, commit: bool = False)

Adds new step for multi-step plugin.

Parameters:

  • href (str) – The URL of the REST entry point resource.

  • ui_href (str) – The URL of the micro frontend that corresponds to the REST entry point resource.

  • step_id (str) – ID of step, e.g., "step1" or "step2b", is automatically appended to previous step

Raises:

AssertionError – raised in case the previous step was not cleared before this method is called.

add_task_log_entry(task_log: str, commit: bool = False)

Appends task_log separated by a new line.

Parameters:

task_log (str) – new entry to be added

clear_previous_step(commit: bool = False)

Set "cleared" of previous step to true if available. Note: call before calling add_next_step.

current_step: Mapped[int]
data: Mapped[Dict[str, Dict[str, Any] | list | str | float | int | bool | None] | List[Dict[str, Any] | list | str | float | int | bool | None] | str | float | int | bool | None]
finished_at: Mapped[datetime | None]
classmethod get_by_id(id_: int) → ProcessingTask | None

Get the object instance by the object id from the database. (None if not found)

property has_uncleared_step: bool
id: Mapped[int]
property is_finished: bool

Return true if the task has finished either successfully or with an error.

property is_ok: bool

Return true if the task has finished successfully.

multi_step: Mapped[bool]
outputs: Mapped[List[TaskFile]]
parameters: Mapped[str]
progress_start: Mapped[float]
progress_target: Mapped[float]
progress_unit: Mapped[str]
progress_value: Mapped[float]
save(commit: bool = False)

Add this object to the current session and optionally commit the session to persist all objects in the session.

started_at: Mapped[datetime]
property status: str

Return the finished status of the task.

If the task is finished but no task_status was set returns "UNKNOWN".

If the task is not finished returns "PENDING".

Returns:

self.task_status | "UNKNOWN" | "PENDING"

Return type:

str

steps: Mapped[OrderingList[Step]]
subscriptions: Mapped[List[TaskUpdateSubscription]]
task_log: Mapped[str]
task_name: Mapped[str]
task_status: Mapped[str | None]
class qhana_plugin_runner.db.models.tasks.Step(id, step_id, href, ui_href, cleared=False)

Bases: object

Step for multi-step plugins

id

ID of corresponding ProcessingTask entry. Use the id to fetch this information from the database.

Type:

int, optional

step_id

ID of step, e.g., "step1" or "step1.step2b".

Type:

str

href

The URL of the REST entry point resource.

Type:

str

ui_href

The URL of the micro frontend that corresponds to the REST entry point resource.

Type:

str

cleared

false if step is awaiting input, only last step in list can be marked as false.

Type:

bool

cleared: Mapped[bool]
href: Mapped[str]
id: Mapped[int]
number: Mapped[int]
step_id: Mapped[str]
ui_href: Mapped[str]
class qhana_plugin_runner.db.models.tasks.TaskFile(task: ProcessingTask, security_tag: str, storage_provider: str, file_name: str, file_storage_data: str, file_type: str | None, mimetype: str | None = None, created_at: datetime = datetime.datetime(2026, 6, 3, 8, 11, 23, 787451))

Bases: object

created_at: Mapped[datetime]
file_name: Mapped[str]
file_storage_data: Mapped[str]
file_type: Mapped[str | None]
classmethod get_by_id(id_: int) → TaskFile | None

Get the object instance by the object id from the database. (None if not found)

classmethod get_task_result_files(task: int | ProcessingTask) → Sequence[TaskFile]

Get a sequence of task result files (e.g. all files with a file-type that is not “temp-file”).

id: Mapped[int]
mimetype: Mapped[str | None]
save(commit: bool = False)

Add this object to the current session and optionally commit the session to persist all objects in the session.

security_tag: Mapped[str]
storage_provider: Mapped[str]
task: Mapped[ProcessingTask]
task_id: Mapped[int | None]

Bases: object

Links that get exposed in the task result.

Use these links to expose additional endpoints relevant for plugin-to-plugin interaction that depends on the current task context.

href: Mapped[str]
id: Mapped[int]
task: Mapped[ProcessingTask]
task_id: Mapped[int | None]
type: Mapped[str]
class qhana_plugin_runner.db.models.tasks.TaskUpdateSubscription(task, webhook_href, task_href, event_type=None)

Bases: object

Table for webhook subscriptions to task updates.

event_type: Mapped[str | None]
classmethod get_by_task_and_event(task: int | ProcessingTask, event: str | None = None) → List[WebhookRef]

Get all webhooks for a task matching the given event type. (‘None’ matches all types.)

classmethod get_by_task_and_subscriber(task: int | ProcessingTask, webhook_href: str, event: str | None = Ellipsis) → List[TaskUpdateSubscription]

Get all webhooks for a task matching the given event type. (‘None’ matches all types.)

id: Mapped[int]
save(commit: bool = False)

Add this object to the current session and optionally commit the session to persist all objects in the session.

task: Mapped[ProcessingTask]
task_href: Mapped[str]
task_id: Mapped[int | None]
webhook_href: Mapped[str]
class qhana_plugin_runner.db.models.tasks.WebhookRef(webhook_href, task_href)

Bases: NamedTuple

task_href: str

Alias for field number 1

webhook_href: str

Alias for field number 0