Migrate the route "/api/invocations/{invocation_id}" of the workflows API to Fast API

client/src/api/schema/schema.ts

@@ -823,6 +823,27 @@ export interface paths {
         /** Prepare history for export-style download and write to supplied URI. */
         post: operations["write_store_api_histories__history_id__write_store_post"];
     };
+    "/api/invocations/{invocation_id}": {
+        /**
+         * Get detailed description of workflow invocation
+         * @description :param  step_details:       fetch details about individual invocation steps
+         *                             and populate a steps attribute in the resulting
+         *                             dictionary. Defaults to false.
+         * :type   step_details:       bool
+         *
+         * :param  legacy_job_state:   If step_details is true, and this is set to true
+         *                             populate the invocation step state with the job state
+         *                             instead of the invocation step state. This will also
+         *                             produce one step per job in mapping jobs to mimic the
+         *                             older behavior with respect to collections. Partially
+         *                             scheduled steps may provide incomplete information
+         *                             and the listed steps outputs are the mapped over
+         *                             step outputs but the individual job outputs
+         *                             when this is set - at least for now.
+         * :type   legacy_job_state:   bool
+         */
+        get: operations["show_invocation_api_invocations__invocation_id__get"];
+    };
     "/api/invocations/{invocation_id}/biocompute": {
         /**
          * Return a BioCompute Object for the workflow invocation.
@@ -14283,6 +14304,59 @@ export interface operations {
             };
         };
     };
+    show_invocation_api_invocations__invocation_id__get: {
+        /**
+         * Get detailed description of workflow invocation
+         * @description :param  step_details:       fetch details about individual invocation steps
+         *                             and populate a steps attribute in the resulting
+         *                             dictionary. Defaults to false.
+         * :type   step_details:       bool
+         *
+         * :param  legacy_job_state:   If step_details is true, and this is set to true
+         *                             populate the invocation step state with the job state
+         *                             instead of the invocation step state. This will also
+         *                             produce one step per job in mapping jobs to mimic the
+         *                             older behavior with respect to collections. Partially
+         *                             scheduled steps may provide incomplete information
+         *                             and the listed steps outputs are the mapped over
+         *                             step outputs but the individual job outputs
+         *                             when this is set - at least for now.
+         * :type   legacy_job_state:   bool
+         */
+        parameters: {
+            /** @description Include details for individual invocation steps. */
+            /**
+             * @deprecated
+             * @description TODO
+             */
+            query?: {
+                step_details?: boolean;
+                legacy_job_state?: boolean;
+            };
+            /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+            header?: {
+                "run-as"?: string;
+            };
+            /** @description The encoded database identifier of the Invocation. */
+            path: {
+                invocation_id: string;
+            };
+        };
+        responses: {
+            /** @description Successful Response */
+            200: {
+                content: {
+                    "application/json": Record<string, never>;
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
     export_invocation_bco_api_invocations__invocation_id__biocompute_get: {
         /**
          * Return a BioCompute Object for the workflow invocation.

lib/galaxy/schema/invocation.py

@@ -1,7 +1,10 @@
+from datetime import datetime
 from enum import Enum
 from typing import (
     Any,
+    Dict,
     Generic,
+    List,
     Optional,
     TypeVar,
     Union,
@@ -10,6 +13,7 @@
 from pydantic import (
     BaseModel,
     Field,
+    Required,
 )
 from pydantic.generics import GenericModel
 from pydantic.utils import GetterDict
@@ -19,6 +23,12 @@
 )
 
 from galaxy.schema.fields import EncodedDatabaseIdField
+from galaxy.schema.schema import (
+    CreateTimeField,
+    EntityIdField,
+    UpdateTimeField,
+    WorkflowIdField,
+)
 
 
 class WarningReason(str, Enum):
@@ -153,6 +163,47 @@ class GenericInvocationEvaluationWarningWorkflowOutputNotFound(
     )
 
 
+# TODO - This already exists in the WorkflowInvocation(lib/galaxy/model/__init__.py).
+# How can I access it to use it here?
+class InvocationState(str, Enum):
+    NEW = "new"  # Brand new workflow invocation... maybe this should be same as READY
+    READY = "ready"  # Workflow ready for another iteration of scheduling.
+    SCHEDULED = "scheduled"  # Workflow has been scheduled.
+    CANCELLED = "cancelled"
+    CANCELLING = "cancelling"  # invocation scheduler will cancel job in next iteration
+    FAILED = "failed"
+
+
+class EncodedInvocation(BaseModel):
+    id: EncodedDatabaseIdField = EntityIdField
+    create_time: datetime = CreateTimeField
+    update_time: datetime = UpdateTimeField
+    workflow_id: EncodedDatabaseIdField = WorkflowIdField
+    # history_id: EncodedDatabaseIdField = HistoryIdField # TODO - fix HistoryIdField?
+    history_id: EncodedDatabaseIdField = Field(
+        default=Required, title="History ID", description="The encoded ID of the history associated with this item."
+    )
+    # TODO this fails weirdly ... try to understand and fix it
+    # uuid: UUID4 = UuidField
+    uuid: Any = Field(..., title="UUID", description="Universal unique identifier for this dataset.")
+    state: InvocationState = Field(
+        default=Required, title="Invocation state", description="State of workflow invocation."
+    )
+    # TODO - is there a class which classifies these models
+    model_class: str = Field(default=Required, title="Model class", description="Model class name.")
+    # TODO - Add proper models
+    steps: Optional[List[Dict[str, Optional[str]]]] = None
+    inputs: Optional[Dict[str, Dict[str, Optional[str]]]] = None
+    input_step_parameters: Optional[Dict[str, Dict[str, Optional[str]]]] = None
+    outputs: Optional[Dict[str, Dict[str, Optional[str]]]] = None
+    output_collections: Optional[Dict[str, Dict[str, Optional[str]]]] = None
+    output_values: Optional[Dict[str, Optional[str]]] = None
+    # TODO understand where this comes from
+    message: Optional[str] = Field(
+        default=None, title="Message", description="Message associated with this invocation."
+    )
+
+
 InvocationCancellationReviewFailed = GenericInvocationCancellationReviewFailed[int]
 InvocationCancellationHistoryDeleted = GenericInvocationCancellationHistoryDeleted[int]
 InvocationCancellationUserRequest = GenericInvocationCancellationUserRequest[int]

lib/galaxy/schema/schema.py

@@ -143,6 +143,10 @@ class DatasetCollectionPopulatedState(str, Enum):
     description="The encoded ID of this entity.",
 )
 
+WorkflowIdField = Field(
+    ..., title="Workflow ID", description="The encoded ID of the workflow that will be run on this step."
+)
+
 DatasetStateField: DatasetState = Field(
     ...,
     title="State",

lib/galaxy/webapps/galaxy/api/workflows.py

@@ -24,6 +24,7 @@
 from markupsafe import escape
 from pydantic import Extra
 from starlette.responses import StreamingResponse
+from typing_extensions import Annotated
 
 from galaxy import (
     exceptions,
@@ -49,7 +50,10 @@
 from galaxy.model.item_attrs import UsesAnnotations
 from galaxy.model.store import BcoExportOptions
 from galaxy.schema.fields import DecodedDatabaseIdField
-from galaxy.schema.invocation import InvocationMessageResponseModel
+from galaxy.schema.invocation import (
+    EncodedInvocation,
+    InvocationMessageResponseModel,
+)
 from galaxy.schema.schema import (
     AsyncFile,
     AsyncTaskResultSummary,
@@ -1198,6 +1202,12 @@ def __encode_invocation(self, invocation, **kwd):
     description="Set this to true to skip joining workflow step counts and optimize the resulting index query. Response objects will not contain step counts.",
 )
 
+StepDetailsQueryParam: bool = Query(
+    title="Include step details", description="Include details for individual invocation steps."
+)
+
+LegacyJobStateQueryParam: bool = Query(title="TODO", description="TODO", deprecated=True)
+
 
 @router.cbv
 class FastAPIWorkflows:
@@ -1503,6 +1513,47 @@ def download_invocation_bco(
             },
         )
 
+    @router.get(
+        "/api/invocations/{invocation_id}",
+        name="show_invocation",
+        summary="Get detailed description of workflow invocation",
+    )
+    def show_invocation(
+        self,
+        invocation_id: Annotated[DecodedDatabaseIdField, InvocationIDPathParam],
+        # view: Annotated[Optional[InvocationSerializationView], Query()] = None,
+        step_details: Annotated[bool, StepDetailsQueryParam] = False,
+        legacy_job_state: Annotated[bool, LegacyJobStateQueryParam] = False,
+        trans: ProvidesUserContext = DependsOnTrans,
+    ):
+        """
+        :param  step_details:       fetch details about individual invocation steps
+                                    and populate a steps attribute in the resulting
+                                    dictionary. Defaults to false.
+        :type   step_details:       bool
+
+        :param  legacy_job_state:   If step_details is true, and this is set to true
+                                    populate the invocation step state with the job state
+                                    instead of the invocation step state. This will also
+                                    produce one step per job in mapping jobs to mimic the
+                                    older behavior with respect to collections. Partially
+                                    scheduled steps may provide incomplete information
+                                    and the listed steps outputs are the mapped over
+                                    step outputs but the individual job outputs
+                                    when this is set - at least for now.
+        :type   legacy_job_state:   bool
+        """
+        workflow_invocation = self.invocations_service._workflows_manager.get_invocation(
+            trans, invocation_id, eager=True
+        )
+        if not workflow_invocation:
+            raise exceptions.ObjectNotFound()
+        serialized_invocation = self.invocations_service.serialize_workflow_invocation(
+            workflow_invocation,
+            InvocationSerializationParams(step_details=step_details, legacy_job_state=legacy_job_state),
+        )
+        return EncodedInvocation(**serialized_invocation)
+
     # TODO: remove this after 23.1 release
     def _deprecated_generate_bco(
         self, trans, invocation_id: DecodedDatabaseIdField, merge_history_metadata: Optional[bool]