qhana_plugin_runner.api.tasks_api module

Module containing endpoints related to task progress and results.

class qhana_plugin_runner.api.tasks_api.OutputDataMetadata(data_type: str, content_type: str, href: str, name: str | None = None)

Bases: object

content_type: str

data_type: str

href: str

name: str | None = None

class qhana_plugin_runner.api.tasks_api.OutputDataMetadataSchema(*, only: Sequence[str] | AbstractSet[str] | None = None, exclude: Sequence[str] | AbstractSet[str] = (), many: bool = False, context: dict | None = None, load_only: Sequence[str] | AbstractSet[str] = (), dump_only: Sequence[str] | AbstractSet[str] = (), partial: bool | Sequence[str] | AbstractSet[str] | None = None, unknown: str | None = None)

Bases: MaBaseSchema

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

remove_empty_attributes(data: Dict[str, Any], **kwargs)

Remove name if it is none.

class qhana_plugin_runner.api.tasks_api.SubscriptionDataSchema(*, only: Sequence[str] | AbstractSet[str] | None = None, exclude: Sequence[str] | AbstractSet[str] = (), many: bool = False, context: dict | None = None, load_only: Sequence[str] | AbstractSet[str] = (), dump_only: Sequence[str] | AbstractSet[str] = (), partial: bool | Sequence[str] | AbstractSet[str] | None = None, unknown: str | None = None)

Bases: MaBaseSchema

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

class qhana_plugin_runner.api.tasks_api.TaskData(status: str, log: Optional[str] = None, progress: Optional[qhana_plugin_runner.api.plugin_schemas.ProgressMetadata] = None, steps: Sequence[qhana_plugin_runner.api.plugin_schemas.StepMetadata] = <factory>, outputs: Sequence[qhana_plugin_runner.api.tasks_api.OutputDataMetadata] = <factory>, links: Sequence[qhana_plugin_runner.api.tasks_api.TaskLink] = <factory>)

Bases: object

links: Sequence[TaskLink]

log: str | None = None

outputs: Sequence[OutputDataMetadata]

progress: ProgressMetadata | None = None

status: str

steps: Sequence[StepMetadata]

class qhana_plugin_runner.api.tasks_api.TaskLink(href: str, type: str)

Bases: object

href: str

type: str

class qhana_plugin_runner.api.tasks_api.TaskLinkSchema(*, only: Sequence[str] | AbstractSet[str] | None = None, exclude: Sequence[str] | AbstractSet[str] = (), many: bool = False, context: dict | None = None, load_only: Sequence[str] | AbstractSet[str] = (), dump_only: Sequence[str] | AbstractSet[str] = (), partial: bool | Sequence[str] | AbstractSet[str] | None = None, unknown: str | None = None)

Bases: MaBaseSchema

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

class qhana_plugin_runner.api.tasks_api.TaskStatusSchema(*, only: Sequence[str] | AbstractSet[str] | None = None, exclude: Sequence[str] | AbstractSet[str] = (), many: bool = False, context: dict | None = None, load_only: Sequence[str] | AbstractSet[str] = (), dump_only: Sequence[str] | AbstractSet[str] = (), partial: bool | Sequence[str] | AbstractSet[str] | None = None, unknown: str | None = None)

Bases: MaBaseSchema

make_task_data(data: Dict[str, Any], **kwargs)

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

remove_empty_attributes(data: Dict[str, Any], **kwargs)

Remove result attributes from serialized tasks that have not finished.

class qhana_plugin_runner.api.tasks_api.TaskView

Bases: MethodView

Task status resource.

convert_task_data(task_data: ProcessingTask)

get(task_id: int)

Get the current task status.

methods: t.ClassVar[t.Collection[str] | None] = {'GET', 'POST'}

The methods this view is registered for. Uses the same default (["GET", "HEAD", "OPTIONS"]) as route and add_url_rule by default.

post(command, task_id: int)

Subscribe to future task status updates.

subscribe(task_data: ProcessingTask, subscription_data)

unsubscribe(task_data: ProcessingTask, subscription_data)