feature: Add support for listing executions from pipeline (aws#930)

Author: nishkris

src/sagemaker/workflow/pipeline.py

@@ -396,6 +396,46 @@ def _interpolate_step_collection_name_in_depends_on(self, step_requests: list):
                 )
                 self._interpolate_step_collection_name_in_depends_on(sub_step_requests)
 
+    def list_executions(
+        self,
+        sort_by: str = None,
+        sort_order: str = None,
+        max_results: int = None,
+        next_token: str = None,
+    ) -> Dict[str, Any]:
+        """Lists a pipeline's executions.
+
+        Args:
+            sort_by (str): The field by which to sort results(CreationTime/PipelineExecutionArn).
+            sort_order (str): The sort order for results (Ascending/Descending).
+            max_results (int): The maximum number of pipeline executions to return in the response.
+            next_token (str):  If the result of the previous ListPipelineExecutions request was
+                truncated, the response includes a NextToken. To retrieve the next set of pipeline
+                executions, use the token in the next request.
+
+        Returns:
+            List of Pipeline Execution Summaries. See
+            boto3 client list_pipeline_executions
+            https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/sagemaker.html#SageMaker.Client.list_pipeline_executions
+        """
+        kwargs = dict(PipelineName=self.name)
+        update_args(
+            kwargs,
+            SortBy=sort_by,
+            SortOrder=sort_order,
+            NextToken=next_token,
+            MaxResults=max_results,
+        )
+        response = self.sagemaker_session.sagemaker_client.list_pipeline_executions(**kwargs)
+
+        # Return only PipelineExecutionSummaries and NextToken from the list_pipeline_executions
+        # response
+        return {
+            key: response[key]
+            for key in ["PipelineExecutionSummaries", "NextToken"]
+            if key in response
+        }
+
 
 def format_start_parameters(parameters: Dict[str, Any]) -> List[Dict[str, Any]]:
     """Formats start parameter overrides as a list of dicts.

tests/unit/sagemaker/workflow/test_pipeline.py

@@ -610,6 +610,31 @@ def test_pipeline_disable_experiment_config():
     )
 
 
+def test_pipeline_list_executions(sagemaker_session_mock):
+    sagemaker_session_mock.sagemaker_client.list_pipeline_executions.return_value = {
+        "PipelineExecutionSummaries": [Mock()],
+        "ResponseMetadata": "metadata",
+    }
+    pipeline = Pipeline(
+        name="MyPipeline",
+        parameters=[ParameterString("alpha", "beta"), ParameterString("gamma", "delta")],
+        steps=[],
+        sagemaker_session=sagemaker_session_mock,
+    )
+    executions = pipeline.list_executions()
+    assert len(executions) == 1
+    assert len(executions["PipelineExecutionSummaries"]) == 1
+    sagemaker_session_mock.sagemaker_client.list_pipeline_executions.return_value = {
+        "PipelineExecutionSummaries": [Mock(), Mock()],
+        "NextToken": "token",
+        "ResponseMetadata": "metadata",
+    }
+    executions = pipeline.list_executions()
+    assert len(executions) == 2
+    assert len(executions["PipelineExecutionSummaries"]) == 2
+    assert executions["NextToken"] == "token"
+
+
 def test_pipeline_execution_basics(sagemaker_session_mock):
     sagemaker_session_mock.sagemaker_client.start_pipeline_execution.return_value = {
         "PipelineExecutionArn": "my:arn"