Allow filtering runtime jobs by program ID

qiskit_ibm/api/clients/runtime.py

@@ -174,19 +174,26 @@ def job_get(self, job_id: str) -> Dict:
         logger.debug("Runtime job get response: %s", response)
         return response
 
-    def jobs_get(self, limit: int = None, skip: int = None, pending: bool = None) -> Dict:
+    def jobs_get(
+            self,
+            limit: int = None,
+            skip: int = None,
+            pending: bool = None,
+            program_id: str = None
+    ) -> Dict:
         """Get job data for all jobs.
 
         Args:
             limit: Number of results to return.
             skip: Number of results to skip.
             pending: Returns 'QUEUED' and 'RUNNING' jobs if True,
                 returns 'DONE', 'CANCELLED' and 'ERROR' jobs if False.
+            program_id: Filter by Program ID.
 
         Returns:
             JSON response.
         """
-        return self.api.jobs_get(limit=limit, skip=skip, pending=pending)
+        return self.api.jobs_get(limit=limit, skip=skip, pending=pending, program_id=program_id)
 
     def job_results(self, job_id: str) -> str:
         """Get the results of a program job.

qiskit_ibm/api/rest/runtime.py

@@ -134,14 +134,21 @@ def program_run(
         data = json.dumps(payload, cls=RuntimeEncoder)
         return self.session.post(url, data=data).json()
 
-    def jobs_get(self, limit: int = None, skip: int = None, pending: bool = None) -> Dict:
+    def jobs_get(
+            self,
+            limit: int = None,
+            skip: int = None,
+            pending: bool = None,
+            program_id: str = None
+    ) -> Dict:
         """Get a list of job data.
 
         Args:
             limit: Number of results to return.
             skip: Number of results to skip.
             pending: Returns 'QUEUED' and 'RUNNING' jobs if True,
                 returns 'DONE', 'CANCELLED' and 'ERROR' jobs if False.
+            program_id: Filter by Program ID.
 
         Returns:
             JSON response.
@@ -154,6 +161,8 @@ def jobs_get(self, limit: int = None, skip: int = None, pending: bool = None) ->
             payload['offset'] = skip
         if pending is not None:
             payload['pending'] = 'true' if pending else 'false'
+        if program_id:
+            payload['program'] = program_id
         return self.session.get(url, params=payload).json()
 
     def logout(self) -> None:

qiskit_ibm/runtime/ibm_runtime_service.py

@@ -12,7 +12,6 @@
 
 """Qiskit runtime service."""
 
-import base64
 import logging
 from typing import Dict, Callable, Optional, Union, List, Any, Type
 import json
@@ -516,7 +515,8 @@ def jobs(
             self,
             limit: Optional[int] = 10,
             skip: int = 0,
-            pending: bool = None
+            pending: bool = None,
+            program_id: str = None
     ) -> List[RuntimeJob]:
         """Retrieve all runtime jobs, subject to optional filtering.
 
@@ -526,6 +526,7 @@ def jobs(
             pending: Filter by job pending state. If ``True``, 'QUEUED' and 'RUNNING'
                 jobs are included. If ``False``, 'DONE', 'CANCELLED' and 'ERROR' jobs
                 are included.
+            program_id: Filter by Program ID.
 
         Returns:
             A list of runtime jobs.
@@ -538,7 +539,8 @@ def jobs(
             jobs_response = self._api_client.jobs_get(
                 limit=current_page_limit,
                 skip=offset,
-                pending=pending)
+                pending=pending,
+                program_id=program_id)
             job_page = jobs_response["jobs"]
             # count is the total number of jobs that would be returned if
             # there was no limit or skip

releasenotes/notes/feature-filter-jobs-by-program-id-e7ef435bed1081be.yaml

@@ -0,0 +1,5 @@
+---
+features:
+  - |
+    You can now pass ``program_id`` parameter to :meth:`qiskit_ibm.runtime.IBMRuntimeService.jobs`
+    method to filter jobs by Program ID.

test/ibm/runtime/fake_runtime_client.py

@@ -315,7 +315,7 @@ def job_get(self, job_id):
         """Get the specific job."""
         return self._get_job(job_id).to_dict()
 
-    def jobs_get(self, limit=None, skip=None, pending=None):
+    def jobs_get(self, limit=None, skip=None, pending=None, program_id=None):
         """Get all jobs."""
         pending_statuses = ['QUEUED', 'RUNNING']
         returned_statuses = ['COMPLETED', 'FAILED', 'CANCELLED']
@@ -327,6 +327,9 @@ def jobs_get(self, limit=None, skip=None, pending=None):
             job_status_list = pending_statuses if pending else returned_statuses
             jobs = [job for job in jobs if job._status in job_status_list]
             count = len(jobs)
+        if program_id:
+            jobs = [job for job in jobs if job._program_id == program_id]
+            count = len(jobs)
         jobs = jobs[skip:limit+skip]
         return {"jobs": [job.to_dict() for job in jobs],
                 "count": count}

test/ibm/runtime/test_runtime.py

@@ -601,6 +601,18 @@ def test_jobs_limit_skip_returned(self):
         rjobs = self.runtime.jobs(limit=limit, skip=skip, pending=False)
         self.assertEqual(limit, len(rjobs))
 
+    def test_jobs_filter_by_program_id(self):
+        """Test retrieving jobs by Program ID."""
+        program_id = self._upload_program()
+        program_id_1 = self._upload_program()
+        job = self._run_program(program_id=program_id)
+        job_1 = self._run_program(program_id=program_id_1)
+        job.wait_for_final_state()
+        job_1.wait_for_final_state()
+        rjobs = self.runtime.jobs(program_id=program_id)
+        self.assertEqual(program_id, rjobs[0].program_id)
+        self.assertEqual(1, len(rjobs))
+
     def test_cancel_job(self):
         """Test canceling a job."""
         job = self._run_program(job_classes=CancelableRuntimeJob)

test/ibm/runtime/test_runtime_integration.py

@@ -376,6 +376,15 @@ def test_retrieve_returned_jobs(self):
                 break
         self.assertTrue(found, f"Returned job {job.job_id} not retrieved.")
 
+    def test_retrieve_jobs_by_program_id(self):
+        """Test retrieving jobs by Program ID."""
+        program_id = self._upload_program()
+        job = self._run_program(program_id=program_id)
+        job.wait_for_final_state()
+        rjobs = self.provider.runtime.jobs(program_id=program_id)
+        self.assertEqual(program_id, rjobs[0].program_id)
+        self.assertEqual(1, len(rjobs))
+
     def test_cancel_job_queued(self):
         """Test canceling a queued job."""
         _ = self._run_program(iterations=10)