Repository: Qiskit/qiskit-ibm-runtime (python)

Issue:
Filter jobs by session id
**What is the expected feature or enhancement?**
The API now supports filtering jobs by session id https://us-east.quantum-computing.cloud.ibm.com/openapi#/Jobs/list_jobs.
Add `session_id` string parameter to `QiskitRuntimeService.jobs()` to be able to filter jobs by `session_id`.

Pull request description:
Filter jobs by session id 
<!--
⚠️ The pull request title should be short detailed and understandable for all.
⚠️ Also, please add a release note file using reno if the change needs to be
  documented in the release notes.
⚠️ If your pull request fixes an open issue, please link to the issue.

✅ I have added the tests to cover my changes.
✅ I have updated the documentation accordingly.
✅ I have read the CONTRIBUTING document.
-->

### Summary

Includes minor update to the release note from https://github.com/Qiskit/qiskit-ibm-runtime/pull/352

### Details and comments
Fixes #355 



Patch:
diff --git a/qiskit_ibm_runtime/api/clients/runtime.py b/qiskit_ibm_runtime/api/clients/runtime.py
index 9a1f411b..780711ca 100644
--- a/qiskit_ibm_runtime/api/clients/runtime.py
+++ b/qiskit_ibm_runtime/api/clients/runtime.py
@@ -218,6 +218,7 @@ class RuntimeClient(BaseBackendClient):
         group: str = None,
         project: str = None,
         job_tags: Optional[List[str]] = None,
+        session_id: Optional[str] = None,
     ) -> Dict:
         """Get job data for all jobs.
 
@@ -231,6 +232,7 @@ class RuntimeClient(BaseBackendClient):
             group: Filter by group - hub, group, and project must all be specified.
             project: Filter by project - hub, group, and project must all be specified.
             job_tags: Filter by tags assigned to jobs. Matched jobs are associated with all tags.
+            session_id: Job ID of the first job in a runtime session.
 
         Returns:
             JSON response.
@@ -244,6 +246,7 @@ class RuntimeClient(BaseBackendClient):
             group=group,
             project=project,
             job_tags=job_tags,
+            session_id=session_id,
         )
 
     def job_results(self, job_id: str) -> str:
diff --git a/qiskit_ibm_runtime/api/rest/runtime.py b/qiskit_ibm_runtime/api/rest/runtime.py
index 041748e2..cf9e4ec1 100644
--- a/qiskit_ibm_runtime/api/rest/runtime.py
+++ b/qiskit_ibm_runtime/api/rest/runtime.py
@@ -175,6 +175,7 @@ class Runtime(RestAdapterBase):
         group: str = None,
         project: str = None,
         job_tags: Optional[List[str]] = None,
+        session_id: Optional[str] = None,
     ) -> Dict:
         """Get a list of job data.
 
@@ -188,6 +189,7 @@ class Runtime(RestAdapterBase):
             group: Filter by group - hub, group, and project must all be specified.
             project: Filter by project - hub, group, and project must all be specified.
             job_tags: Filter by tags assigned to jobs. Matched jobs are associated with all tags.
+            session_id: Job ID of the first job in a runtime session.
 
         Returns:
             JSON response.
@@ -204,6 +206,8 @@ class Runtime(RestAdapterBase):
             payload["program"] = program_id
         if job_tags:
             payload["tags"] = job_tags
+        if session_id:
+            payload["session_id"] = session_id
         if all([hub, group, project]):
             payload["provider"] = f"{hub}/{group}/{project}"
         return self.session.get(url, params=payload).json()
diff --git a/qiskit_ibm_runtime/qiskit_runtime_service.py b/qiskit_ibm_runtime/qiskit_runtime_service.py
index 9f3a2619..1d4e65ae 100644
--- a/qiskit_ibm_runtime/qiskit_runtime_service.py
+++ b/qiskit_ibm_runtime/qiskit_runtime_service.py
@@ -1218,6 +1218,7 @@ class QiskitRuntimeService:
         program_id: str = None,
         instance: Optional[str] = None,
         job_tags: Optional[List[str]] = None,
+        session_id: Optional[str] = None,
     ) -> List[RuntimeJob]:
         """Retrieve all runtime jobs, subject to optional filtering.
 
@@ -1231,6 +1232,7 @@ class QiskitRuntimeService:
             instance: This is only supported for ``ibm_quantum`` runtime and is in the
                 hub/group/project format.
             job_tags: Filter by tags assigned to jobs. Matched jobs are associated with all tags.
+            session_id: Job ID of the first job in a runtime session.
 
         Returns:
             A list of runtime jobs.
@@ -1260,6 +1262,7 @@ class QiskitRuntimeService:
                 group=group,
                 project=project,
                 job_tags=job_tags,
+                session_id=session_id,
             )
             job_page = jobs_response["jobs"]
             # count is the total number of jobs that would be returned if
diff --git a/releasenotes/notes/filter-jobs-session-id-e8296c55ad75f789.yaml b/releasenotes/notes/filter-jobs-session-id-e8296c55ad75f789.yaml
new file mode 100644
index 00000000..f763b9d2
--- /dev/null
+++ b/releasenotes/notes/filter-jobs-session-id-e8296c55ad75f789.yaml
@@ -0,0 +1,6 @@
+---
+upgrade:
+  - |
+    The ``session_id``, which is the Job ID of the first job in a runtime session can now be used 
+    as a filter in :meth:`~qiskit_ibm_runtime.QiskitRuntimeService.jobs` with the parameter
+    ``session_id``.
diff --git a/releasenotes/notes/job-tags-support-465d3be18b20b306.yaml b/releasenotes/notes/job-tags-support-465d3be18b20b306.yaml
index c0aa6258..fb1f3480 100644
--- a/releasenotes/notes/job-tags-support-465d3be18b20b306.yaml
+++ b/releasenotes/notes/job-tags-support-465d3be18b20b306.yaml
@@ -1,6 +1,6 @@
 ---
 upgrade:
   - |
-    :meth:`~qiskit_ibm_runtime.IBMRuntimeService.run` now supports a new parameter,
+    :meth:`~qiskit_ibm_runtime.QiskitRuntimeService.run` now supports a new parameter,
     ``job_tags``. These tags can be used when filtering jobs with 
-    :meth:`~qiskit_ibm_runtime.IBMRuntimeService.jobs`.
+    :meth:`~qiskit_ibm_runtime.QiskitRuntimeService.jobs`.


Test patch:
diff --git a/test/ibm_test_case.py b/test/ibm_test_case.py
index 5dfa2d54..a4fcaa5c 100644
--- a/test/ibm_test_case.py
+++ b/test/ibm_test_case.py
@@ -185,6 +185,7 @@ class IBMIntegrationJobTestCase(IBMIntegrationTestCase):
         log_level=None,
         job_tags=None,
         max_execution_time=None,
+        session_id=None,
     ):
         """Run a program."""
         self.log.debug("Running program on %s", service.channel)
@@ -208,6 +209,7 @@ class IBMIntegrationJobTestCase(IBMIntegrationTestCase):
             options=options,
             job_tags=job_tags,
             max_execution_time=max_execution_time,
+            session_id=session_id,
             callback=callback,
         )
         self.log.info("Runtime job %s submitted.", job.job_id)
diff --git a/test/integration/test_retrieve_job.py b/test/integration/test_retrieve_job.py
index e96646fd..9a2c808f 100644
--- a/test/integration/test_retrieve_job.py
+++ b/test/integration/test_retrieve_job.py
@@ -138,6 +138,18 @@ class TestIntegrationRetrieveJob(IBMIntegrationJobTestCase):
         rjobs = service.jobs(job_tags=["no_test_tag"])
         self.assertFalse(rjobs)
 
+    @run_integration_test
+    def test_retrieve_jobs_by_session_id(self, service):
+        """Test retrieving jobs by session_id."""
+        job = self._run_program(service)
+        job.wait_for_final_state()
+        job_2 = self._run_program(service, session_id=job.job_id)
+        job_2.wait_for_final_state()
+        rjobs = service.jobs(session_id=job.job_id)
+        self.assertEqual(2, len(rjobs), f"Retrieved jobs: {[j.job_id for j in rjobs]}")
+        rjobs = service.jobs(session_id="test")
+        self.assertFalse(rjobs)
+
     @run_integration_test
     def test_jobs_filter_by_hgp(self, service):
         """Test retrieving jobs by hgp."""
diff --git a/test/program.py b/test/program.py
index a1a581c5..1ec06f21 100644
--- a/test/program.py
+++ b/test/program.py
@@ -81,6 +81,7 @@ def run_program(
     log_level=None,
     job_tags=None,
     max_execution_time=None,
+    session_id=None,
 ):
     """Run a program."""
     backend_name = backend_name if backend_name is not None else "common_backend"
@@ -99,5 +100,6 @@ def run_program(
         instance=instance,
         job_tags=job_tags,
         max_execution_time=max_execution_time,
+        session_id=session_id,
     )
     return job
diff --git a/test/unit/mock/fake_runtime_client.py b/test/unit/mock/fake_runtime_client.py
index b9bc6260..1e1384cc 100644
--- a/test/unit/mock/fake_runtime_client.py
+++ b/test/unit/mock/fake_runtime_client.py
@@ -379,6 +379,9 @@ class BaseFakeRuntimeClient:
         if backend_name is None:
             backend_name = self.list_backends()[0]
 
+        if session_id is None:
+            session_id = job_id
+
         job = job_cls(
             job_id=job_id,
             program_id=program_id,
@@ -417,6 +420,7 @@ class BaseFakeRuntimeClient:
         group=None,
         project=None,
         job_tags=None,
+        session_id=None,
     ):
         """Get all jobs."""
         pending_statuses = ["QUEUED", "RUNNING"]
@@ -442,6 +446,9 @@ class BaseFakeRuntimeClient:
         if job_tags:
             jobs = [job for job in jobs if job._job_tags == job_tags]
             count = len(jobs)
+        if session_id:
+            jobs = [job for job in jobs if job._session_id == session_id]
+            count = len(jobs)
         jobs = jobs[skip : limit + skip]
         return {"jobs": [job.to_dict() for job in jobs], "count": count}
 
diff --git a/test/unit/test_job_retrieval.py b/test/unit/test_job_retrieval.py
index d76b2054..cf9f9667 100644
--- a/test/unit/test_job_retrieval.py
+++ b/test/unit/test_job_retrieval.py
@@ -225,6 +225,24 @@ class TestRetrieveJobs(IBMTestCase):
         rjobs = service.jobs(program_id=program_id, job_tags=["no_test_tag"])
         self.assertFalse(rjobs)
 
+    def test_jobs_filter_by_session_id(self):
+        """Test retrieving jobs by session id."""
+        service = self._ibm_quantum_service
+        program_id = upload_program(service)
+
+        job = run_program(service=service, program_id=program_id)
+        job_2 = run_program(
+            service=service, program_id=program_id, session_id=job.job_id
+        )
+        with mock_wait_for_final_state(service, job):
+            job.wait_for_final_state()
+            job_2.wait_for_final_state()
+        rjobs = service.jobs(program_id=program_id, session_id=job.job_id)
+        self.assertTrue(rjobs)
+        self.assertEqual(2, len(rjobs))
+        rjobs = service.jobs(program_id=program_id, session_id="no_test_session_id")
+        self.assertFalse(rjobs)
+
     def test_jobs_bad_instance(self):
         """Test retrieving jobs with bad instance values."""
         service = self._ibm_quantum_service