feat: add PipelineJob.list

google/cloud/aiplatform/pipeline_jobs.py

@@ -18,7 +18,7 @@
 import datetime
 import time
 import re
-from typing import Any, Optional, Dict
+from typing import Any, Dict, List, Optional
 
 from google.auth import credentials as auth_credentials
 from google.cloud.aiplatform import base
@@ -376,6 +376,54 @@ def cancel(self) -> None:
         """
         self.api_client.cancel_pipeline_job(name=self.resource_name)
 
+    @classmethod
+    def list(
+        cls,
+        filter: Optional[str] = None,
+        order_by: Optional[str] = None,
+        project: Optional[str] = None,
+        location: Optional[str] = None,
+        credentials: Optional[auth_credentials.Credentials] = None,
+    ) -> List["PipelineJob"]:
+        """List all instances of this PipelineJob resource.
+
+        Example Usage:
+
+        aiplatform.PipelineJob.list(
+            filter='display_name="experiment_a27"',
+            order_by='create_time desc'
+        )
+
+        Args:
+            filter (str):
+                Optional. An expression for filtering the results of the request.
+                For field names both snake_case and camelCase are supported.
+            order_by (str):
+                Optional. A comma-separated list of fields to order by, sorted in
+                ascending order. Use "desc" after a field name for descending.
+                Supported fields: `display_name`, `create_time`, `update_time`
+            project (str):
+                Optional. Project to retrieve list from. If not set, project
+                set in aiplatform.init will be used.
+            location (str):
+                Optional. Location to retrieve list from. If not set, location
+                set in aiplatform.init will be used.
+            credentials (auth_credentials.Credentials):
+                Optional. Custom credentials to use to retrieve list. Overrides
+                credentials set in aiplatform.init.
+
+        Returns:
+            List[PipelineJob] - A list of PipelineJob resource objects
+        """
+
+        return cls._list_with_local_order(
+            filter=filter,
+            order_by=order_by,
+            project=project,
+            location=location,
+            credentials=credentials,
+        )
+
     def wait_for_resource_creation(self) -> None:
         """Waits until resource has been created."""
         self._wait_for_resource_creation()

tests/unit/aiplatform/test_pipeline_jobs.py

@@ -162,6 +162,14 @@ def mock_pipeline_service_cancel():
         yield mock_cancel_pipeline_job
 
 
+@pytest.fixture
+def mock_pipeline_service_list():
+    with mock.patch.object(
+        pipeline_service_client_v1beta1.PipelineServiceClient, "list_pipeline_jobs"
+    ) as mock_list_pipeline_jobs:
+        yield mock_list_pipeline_jobs
+
+
 @pytest.fixture
 def mock_load_json():
     with patch.object(storage.Blob, "download_as_bytes") as mock_load_json:
@@ -278,6 +286,29 @@ def test_cancel_pipeline_job(
             name=_TEST_PIPELINE_JOB_NAME
         )
 
+    @pytest.mark.usefixtures(
+        "mock_pipeline_service_create", "mock_pipeline_service_get", "mock_load_json",
+    )
+    def test_list_pipeline_job(self, mock_pipeline_service_list):
+        aiplatform.init(
+            project=_TEST_PROJECT,
+            staging_bucket=_TEST_GCS_BUCKET_NAME,
+            credentials=_TEST_CREDENTIALS,
+        )
+
+        job = pipeline_jobs.PipelineJob(
+            display_name=_TEST_PIPELINE_JOB_DISPLAY_NAME,
+            template_path=_TEST_TEMPLATE_PATH,
+            job_id=_TEST_PIPELINE_JOB_ID,
+        )
+
+        job.run()
+        job.list()
+
+        mock_pipeline_service_list.assert_called_once_with(
+            request={"parent": _TEST_PARENT, "filter": None}
+        )
+
     @pytest.mark.usefixtures(
         "mock_pipeline_service_create", "mock_pipeline_service_get", "mock_load_json",
     )