Add a FlowsClient.get_run_definition() method

changelog.d/20230809_125026_kurtmckee_get_run_definitions_sc_25840.rst

@@ -0,0 +1,4 @@
+Added
+~~~~~
+
+- Add a ``FlowsClient.get_run_definition()`` method. (:pr:`NUMBER`)

src/globus_sdk/_testing/data/flows/get_run_definition.py

@@ -0,0 +1,27 @@
+import uuid
+
+from globus_sdk._testing import RegisteredResponse, ResponseList, ResponseSet
+
+from ._common import RUN_ID
+
+GET_RUN_DEFINITION = {
+    "flow_id": str(uuid.uuid4()),
+    "definition": {
+        "States": {"no-op": {"End": True, "Type": "Pass"}},
+        "StartAt": "no-op",
+    },
+    "input_schema": {},
+}
+
+
+RESPONSES = ResponseSet(
+    metadata={"run_id": RUN_ID},
+    default=ResponseList(
+        RegisteredResponse(
+            service="flows",
+            method="GET",
+            path=f"/runs/{RUN_ID}/definition",
+            json=GET_RUN_DEFINITION,
+        ),
+    ),
+)

src/globus_sdk/services/flows/client.py

@@ -645,6 +645,40 @@ def get_run(
 
         return self.get(f"/runs/{run_id}", query_params=query_params)
 
+    def get_run_definition(
+        self,
+        run_id: UUIDLike,
+    ) -> GlobusHTTPResponse:
+        """
+        Get the flow definition and input schema at the time the run was started.
+
+        :param run_id: The ID of the run to get
+        :type run_id: str or UUID
+
+        .. tab-set::
+
+            .. tab-item:: Example Usage
+
+                .. code-block:: python
+
+                    from globus_sdk import FlowsClient
+
+                    flows = FlowsClient(...)
+                    flows.get_run_definition("581753c7-45da-43d3-ad73-246b46e7cb6b")
+
+            .. tab-item:: Example Response Data
+
+                .. expandtestfixture:: flows.get_run_definition
+
+            .. tab-item:: API Info
+
+                .. extdoclink:: Get Run Definition
+                    :service: flows
+                    :ref: Flows/paths/~1runs~1{run_id}~1definition/get
+        """
+
+        return self.get(f"/runs/{run_id}/definition")
+
     def cancel_run(self, run_id: UUIDLike) -> GlobusHTTPResponse:
         """
         Cancel a run.

tests/functional/services/flows/test_run_crud.py

@@ -6,6 +6,17 @@
 from globus_sdk._testing import get_last_request, load_response
 
 
+def test_get_run_definition(flows_client):
+    """Validate the HTTP method and route used to get the flow definition for a run."""
+
+    run_id = load_response(flows_client.get_run_definition).metadata["run_id"]
+
+    flows_client.get_run_definition(run_id)
+    request = get_last_request()
+    assert request.method == "GET"
+    assert request.url.endswith(f"/runs/{run_id}/definition")
+
+
 def test_cancel_run(flows_client):
     """Verify that run cancellation requests meet expectations."""