Source code for galaxy.schema.tasks

from enum import Enum
from typing import (
    Literal,
    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: Optional[int] = None
    galaxy_session_id: Optional[int] = None




[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
    export_association_id: Optional[int] = None




[docs]
class WriteInvocationTo(WriteStoreToPayload, BcoGenerationTaskParametersMixin):
    invocation_id: int
    user: RequestUser
    export_association_id: Optional[int] = None




[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 CopyDatasetsPayloadSourceEntry(Model):
    id: str
    type: str




[docs]
class CopyDatasetsPayload(Model):
    source_content: list[CopyDatasetsPayloadSourceEntry]
    target_history_ids: Optional[list[str]] = None
    target_history_name: Optional[str] = None




[docs]
class CopyDatasetsResponse(Model):
    history_ids: list[str]




[docs]
class PurgeDatasetsTaskRequest(Model):
    dataset_ids: list[int]




[docs]
class PurgeHistoryDatasetsTaskRequest(Model):
    history_id: int
    preserve_owner_update_time: bool = False




[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.",
    )



TOOL_SOURCE_CLASS = Literal["XmlToolSource", "YamlToolSource", "CwlToolSource"]



[docs]
class ToolSource(Model):
    raw_tool_source: str
    tool_dir: Optional[str] = None
    tool_source_class: TOOL_SOURCE_CLASS = "XmlToolSource"




[docs]
class QueueJobs(Model):
    tool_source: ToolSource
    tool_request_id: int  # links to request ("incoming") and history
    user: RequestUser  # TODO: test anonymous users through this submission path
    use_cached_jobs: bool
    rerun_remap_job_id: Optional[int]  # link to a job to rerun & remap
    preferred_object_store_id: Optional[str] = None
    tags: Optional[list[str]] = None
    data_manager_mode: Optional[str] = None
    send_email_notification: bool = False
    credentials_context: Optional[list[dict]] = None
    dynamic_tool_id: Optional[int] = None  # link to DynamicTool for custom/user tools