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