Migrate a part of the jobs API to Fast API

lib/galaxy/managers/jobs.py

@@ -880,6 +880,8 @@ def summarize_job_parameters(trans, job):
     Precondition: the caller has verified the job is accessible to the user
     represented by the trans parameter.
     """
+    # More client logic here than is ideal but it is hard to reason about
+    # tool parameter types on the client relative to the server.
 
     def inputs_recursive(input_params, param_values, depth=1, upgrade_messages=None):
         if upgrade_messages is None:

lib/galaxy/schema/jobs.py

@@ -0,0 +1,233 @@
+import json
+from typing import (
+    Any,
+    Dict,
+    List,
+    Optional,
+)
+
+from pydantic import (
+    Extra,
+    Field,
+    Required,
+    UUID4,
+    validator,
+)
+from typing_extensions import Literal
+
+from galaxy.schema.fields import (
+    DecodedDatabaseIdField,
+    EncodedDatabaseIdField,
+)
+from galaxy.schema.schema import (
+    DataItemSourceType,
+    EncodedDataItemSourceId,
+    EntityIdField,
+    JobState,
+    JobSummary,
+    Model,
+)
+
+
+class JobInputSummary(Model):
+    has_empty_inputs: bool = Field(
+        default=Required,
+        title="Empty inputs",
+        description="Job has empty inputs.",
+    )
+    has_duplicate_inputs: bool = Field(
+        default=Required,
+        title="Duplicate inputs",
+        description="Job has duplicate inputs.",
+    )
+
+
+# TODO: Use Tuple again when `make update-client-api-schema` supports them
+class JobErrorSummary(Model):
+    # messages: List[Union[Tuple[str, str], List[str]]]
+    messages: List[List[str]] = Field(
+        default=Required,
+        title="Error messages",
+        description="The error messages for the specified job.",
+    )
+
+
+class JobAssociation(Model):
+    name: str = Field(
+        default=Required,
+        title="name",
+        description="Name of the job parameter.",
+    )
+    dataset: EncodedDataItemSourceId = Field(
+        default=Required,
+        title="dataset",
+        description="Reference to the associated item.",
+    )
+
+
+class JobInputAssociation(JobAssociation):
+    name: str = Field(
+        default=Required,
+        title="name",
+        description="Name of the job input parameter.",
+    )
+
+
+class JobOutputAssociation(JobAssociation):
+    name: str = Field(
+        default=Required,
+        title="name",
+        description="Name of the job output parameter.",
+    )
+
+
+class ReportJobErrorPayload(Model):
+    dataset_id: DecodedDatabaseIdField = Field(
+        default=Required,
+        title="History Dataset Association ID",
+        description="The History Dataset Association ID related to the error.",
+    )
+    email: Optional[str] = Field(
+        default=None,
+        title="Email",
+        description="Email address for communication with the user. Only required for anonymous users.",
+    )
+    message: Optional[str] = Field(
+        default=None,
+        title="Message",
+        description="The optional message sent with the error report.",
+    )
+
+
+class SearchJobsPayload(Model):
+    tool_id: str = Field(
+        default=Required,
+        title="Tool ID",
+        description="The tool ID related to the job.",
+    )
+    inputs: Dict[str, Any] = Field(
+        default=Required,
+        title="Inputs",
+        description="The inputs of the job.",
+    )
+    state: JobState = Field(
+        default=Required,
+        title="State",
+        description="Current state of the job.",
+    )
+
+    @validator("inputs", pre=True)
+    def decode_json(cls, v):
+        if isinstance(v, str):
+            return json.loads(v)
+        return v
+
+    class Config:
+        extra = Extra.allow  # This is used for items named file_ and __file_
+
+
+class DeleteJobPayload(Model):
+    message: Optional[str] = Field(
+        default=None,
+        title="Job message",
+        description="Stop message",
+    )
+
+
+class SrcItem(Model):
+    src: DataItemSourceType
+
+
+class EncodedHdcaSourceId(SrcItem):
+    id: EncodedDatabaseIdField = EntityIdField
+    src: Literal[DataItemSourceType.hdca] = Field(
+        default=Required,
+        title="Source",
+        description="The source of this dataset, which in the case of the model can only be `hdca`.",
+    )
+
+
+class EncodedDatasetJobInfo(EncodedDataItemSourceId):
+    uuid: Optional[UUID4] = Field(
+        default=None,
+        deprecated=True,
+        title="UUID",
+        description="Universal unique identifier for this dataset.",
+    )
+
+
+class EncodedJobDetails(JobSummary):
+    command_version: str = Field(
+        ...,
+        title="Command Version",
+        description="Tool version indicated during job execution.",
+    )
+    params: Any = Field(
+        ...,
+        title="Parameters",
+        description=(
+            "Object containing all the parameters of the tool associated with this job. "
+            "The specific parameters depend on the tool itself."
+        ),
+    )
+    inputs: Dict[str, EncodedDatasetJobInfo] = Field(
+        {},
+        title="Inputs",
+        description="Dictionary mapping all the tool inputs (by name) to the corresponding data references.",
+    )
+    outputs: Dict[str, EncodedDatasetJobInfo] = Field(
+        {},
+        title="Outputs",
+        description="Dictionary mapping all the tool outputs (by name) to the corresponding data references.",
+    )
+    copied_from_job_id: Optional[EncodedDatabaseIdField] = Field(
+        default=None, title="Copied from Job-ID", description="Reference to cached job if job execution was cached."
+    )
+    output_collections: Dict[str, EncodedHdcaSourceId] = Field(default={}, title="Output collections", description="")
+
+
+class JobDestinationParams(Model):
+    runner: Optional[str] = Field(None, title="Runner", description="Job runner class", alias="Runner")
+    runner_job_id: Optional[str] = Field(
+        None,
+        title="Runner Job ID",
+        description="ID assigned to submitted job by external job running system",
+        alias="Runner Job ID",
+    )
+    handler: Optional[str] = Field(
+        None, title="Handler", description="Name of the process that handled the job.", alias="Handler"
+    )
+
+
+class JobOutput(Model):
+    label: Any = Field(default=Required, title="Output label", description="The output label")  # check if this is true
+    value: EncodedDataItemSourceId = Field(default=Required, title="Dataset", description="The associated dataset.")
+
+
+class JobParameter(Model):
+    text: str = Field(
+        default=Required,
+        title="Text",
+        description="Text associated with the job parameter.",
+    )
+    depth: int = Field(
+        default=Required,
+        title="Depth",
+        description="The depth of the job parameter.",
+    )
+    value: Any = Field(default=Required, title="Value", description="The values of the job parameter")
+    notes: Optional[str] = Field(default=None, title="Notes", description="Notes associated with the job parameter.")
+
+
+class JobDisplayParametersSummary(Model):
+    parameters: List[JobParameter] = Field(
+        default=Required, title="Parameters", description="The parameters of the job in a nested format."
+    )
+    has_parameter_errors: bool = Field(
+        default=Required, title="Has parameter errors", description="The job has parameter errors"
+    )
+    outputs: Dict[str, List[JobOutput]] = Field(
+        default=Required,
+        title="Outputs",
+        description="Dictionary mapping all the tool outputs (by name) with the corresponding dataset information in a nested format.",
+    )

lib/galaxy/schema/schema.py

@@ -257,8 +257,6 @@ class DatasetCollectionPopulatedState(str, Enum):
 
 
 class Model(BaseModel):
-    """Base model definition with common configuration used by all derived models."""
-
     class Config:
         use_enum_values = True  # when using .dict()
         allow_population_by_field_name = True
@@ -475,6 +473,14 @@ class DatasetSourceType(str, Enum):
     ldda = "ldda"
 
 
+class DataItemSourceType(str, Enum):
+    hda = "hda"
+    ldda = "ldda"
+    hdca = "hdca"
+    dce = "dce"
+    dc = "dc"
+
+
 class ColletionSourceType(str, Enum):
     hda = "hda"
     ldda = "ldda"
@@ -1821,18 +1827,18 @@ class JobIdResponse(Model):
 
 
 class JobBaseModel(Model):
-    id: DecodedDatabaseIdField = EntityIdField
+    id: EncodedDatabaseIdField = EntityIdField
+    history_id: Optional[EncodedDatabaseIdField] = Field(
+        None,
+        title="History ID",
+        description="The encoded ID of the history associated with this item.",
+    )
     model_class: JOB_MODEL_CLASS = ModelClassField(JOB_MODEL_CLASS)
     tool_id: str = Field(
         ...,
         title="Tool ID",
         description="Identifier of the tool that generated this job.",
     )
-    history_id: Optional[DecodedDatabaseIdField] = Field(
-        None,
-        title="History ID",
-        description="The encoded ID of the history associated with this item.",
-    )
     state: JobState = Field(
         ...,
         title="State",
@@ -1925,6 +1931,15 @@ class EncodedDatasetSourceId(DatasetSourceIdBase):
     id: EncodedDatabaseIdField = EntityIdField
 
 
+class EncodedDataItemSourceId(Model):
+    id: EncodedDatabaseIdField = EntityIdField
+    src: DataItemSourceType = Field(
+        ...,
+        title="Source",
+        description="The source of this dataset, either `hda`, `ldda`, `hdca`, `dce` or `dc` depending of its origin.",
+    )
+
+
 class DatasetJobInfo(DatasetSourceId):
     uuid: UUID4 = UuidField
 

lib/galaxy/security/idencoding.py

@@ -35,9 +35,11 @@ def __init__(self, **config):
         per_kind_id_secret_base = config.get("per_kind_id_secret_base", self.id_secret)
         self.id_ciphers_for_kind = _cipher_cache(per_kind_id_secret_base)
 
-    def encode_id(self, obj_id, kind=None):
+    def encode_id(self, obj_id, kind=None, strict_integer=False):
         if obj_id is None:
             raise galaxy.exceptions.MalformedId("Attempted to encode None id")
+        if strict_integer and not isinstance(obj_id, int):
+            raise galaxy.exceptions.MalformedId("Attempted to encode id that is not an integer")
         id_cipher = self.__id_cipher(kind)
         # Convert to bytes
         s = smart_str(obj_id)