Source code for galaxy.schema.tasks

from enum import Enum
from typing import (
    List,
    Optional,
)
from uuid import UUID

from pydantic import Field

from galaxy.util.hash_util import HashFunctionNameEnum
from . import PdfDocumentType
from .schema import (
    BcoGenerationParametersMixin,
    DatasetSourceType,
    HistoryContentType,
    Model,
    ModelStoreFormat,
    ShortTermStoreExportPayload,
    WriteStoreToPayload,
)


[docs] class SetupHistoryExportJob(Model): history_id: int job_id: int store_directory: str include_files: bool include_hidden: bool include_deleted: bool
[docs] class PrepareDatasetCollectionDownload(Model): short_term_storage_request_id: UUID history_dataset_collection_association_id: int
[docs] class GeneratePdfDownload(Model): short_term_storage_request_id: UUID # basic markdown - Galaxy directives need to be processed before handing off to this task basic_markdown: str document_type: PdfDocumentType
# serialize user info for tasks
[docs] class RequestUser(Model): user_id: int
# TODO: allow make the above optional and allow a session_id for anonymous users... # session_id: Optional[str]
[docs] class GenerateHistoryDownload(ShortTermStoreExportPayload): history_id: int user: RequestUser export_association_id: Optional[int] = None
[docs] class GenerateHistoryContentDownload(ShortTermStoreExportPayload): content_type: HistoryContentType content_id: int user: RequestUser
[docs] class BcoGenerationTaskParametersMixin(BcoGenerationParametersMixin): galaxy_url: str
[docs] class GenerateInvocationDownload(ShortTermStoreExportPayload, BcoGenerationTaskParametersMixin): invocation_id: int user: RequestUser
[docs] class WriteInvocationTo(WriteStoreToPayload, BcoGenerationTaskParametersMixin): invocation_id: int user: RequestUser
[docs] class WriteHistoryContentTo(WriteStoreToPayload): content_type: HistoryContentType content_id: int user: RequestUser
[docs] class WriteHistoryTo(WriteStoreToPayload): history_id: int user: RequestUser export_association_id: Optional[int] = None
[docs] class ImportModelStoreTaskRequest(Model): user: RequestUser history_id: Optional[int] = None source_uri: str for_library: bool model_store_format: Optional[ModelStoreFormat] = None
[docs] class MaterializeDatasetInstanceTaskRequest(Model): history_id: int user: RequestUser source: DatasetSourceType = Field( title="Source", description="The source of the content. Can be other history element to be copied or library elements.", ) content: int = Field( title="Content", description=( "Depending on the `source` it can be:\n" "- The decoded id of the source library dataset\n" "- The decoded id of the HDA\n" ), )
[docs] class ComputeDatasetHashTaskRequest(Model): dataset_id: int extra_files_path: Optional[str] = None hash_function: HashFunctionNameEnum user: Optional[RequestUser] = None # access checks should be done pre-celery so this is optional
[docs] class PurgeDatasetsTaskRequest(Model): dataset_ids: List[int]
[docs] class TaskState(str, Enum): """Enum representing the possible states of a task.""" PENDING = "PENDING" """The task is waiting for execution.""" STARTED = "STARTED" """The task has been started.""" RETRY = "RETRY" """The task is to be retried, possibly because of failure.""" FAILURE = "FAILURE" """The task raised an exception, or has exceeded the retry limit.""" SUCCESS = "SUCCESS" """The task executed successfully."""
[docs] class TaskResult(Model): """Contains information about the result of an asynchronous task.""" state: TaskState = Field( title="State", description="The current state of the task.", ) result: str = Field( title="Result", description="The result message of the task. Empty if the task is still running. If the task failed, this will contain the exception message.", )