Retrieve interim results from runtime jobs

qiskit_ibm_runtime/api/clients/runtime.py

@@ -216,6 +216,17 @@ def job_results(self, job_id: str) -> str:
             Job result.
         """
         return self.api.program_job(job_id).results()
+
+    def job_interim_results(self, job_id: str) -> str:
+        """Get the interim results of a program job.
+
+        Args:
+            job_id: Program job ID.
+
+        Returns:
+            Job interim results.
+        """
+        return self.api.program_job(job_id).interim_results()
 
     def job_cancel(self, job_id: str) -> None:
         """Cancel a job.

qiskit_ibm_runtime/api/rest/runtime.py

@@ -30,7 +30,8 @@ class Runtime(RestAdapterBase):
     URL_MAP = {
         'programs': '/programs',
         'jobs': '/jobs',
-        'logout': '/logout'
+        'logout': '/logout',
+        'interim_results': '/interim_results'
     }
 
     def program(self, program_id: str) -> 'Program':
@@ -316,6 +317,15 @@ def delete(self) -> None:
         """Delete program job."""
         self.session.delete(self.get_url('self'))
 
+    def interim_results(self) -> str:
+        """Return program job interim results.
+
+        Returns:
+            Interim results.
+        """
+        response = self.session.get(self.get_url('interim_results'))
+        return response.text
+
     def results(self) -> str:
         """Return program job results.
 

qiskit_ibm_runtime/runtime_job.py

@@ -105,13 +105,15 @@ def __init__(
         self._backend = backend
         self._api_client = api_client
         self._results: Optional[Any] = None
+        self._interim_results: Optional[Any] = None
         self._params = params or {}
         self._creation_date = creation_date
         self._program_id = program_id
         self._status = JobStatus.INITIALIZING
         self._error_message = None  # type: Optional[str]
         self._result_decoder = result_decoder
         self._image = image
+        self._final_interim_results = False
 
         # Used for streaming
         self._ws_client_future = None  # type: Optional[futures.Future]
@@ -125,6 +127,26 @@ def __init__(
         if user_callback is not None:
             self.stream_results(user_callback)
 
+    def interim_results(self, decoder: Optional[Type[ResultDecoder]] = None) -> Any:
+        """Return the interim results of the job.
+
+        Args:
+            decoder: A :class:`ResultDecoder` subclass used to decode interim results.
+
+        Returns:
+            Runtime job interim results.
+
+        Raises:
+            RuntimeJobFailureError: If the job failed.
+        """
+        if not self._final_interim_results:
+            _decoder = decoder or self._result_decoder
+            interim_results_raw = self._api_client.job_interim_results(job_id=self.job_id)
+            self._interim_results = _decoder.decode(interim_results_raw)
+            if self.status() in JOB_FINAL_STATES:
+                self._final_interim_results = True
+        return self._interim_results
+
     def result(
             self,
             timeout: Optional[float] = None,

releasenotes/notes/interim-results-b5a18a3784063d56.yaml

@@ -0,0 +1,7 @@
+---
+features:
+  - |
+    You can now use the :meth:`qiskit_ibm_runtime.RuntimeJob.interim_results`
+    method to retrieve runtime program interim results. 
+    Note that interim results will only be available for
+    up to two days.

test/ibm/runtime/fake_runtime_client.py

@@ -89,6 +89,7 @@ def __init__(self, job_id, program_id, hub, group, project, backend_name, final_
         self._backend_name = backend_name
         self._params = params
         self._image = image
+        self._interim_results = json.dumps("foo")
         if final_status is None:
             self._future = self._executor.submit(self._auto_progress)
             self._result = None
@@ -121,6 +122,10 @@ def result(self):
         """Return job result."""
         return self._result
 
+    def interim_results(self):
+        """Return job interim results."""
+        return self._interim_results
+
 
 class FailedRuntimeJob(BaseFakeRuntimeJob):
     """Class for faking a failed runtime job."""
@@ -365,6 +370,10 @@ def set_program_visibility(self, program_id: str, public: bool) -> None:
     def job_results(self, job_id):
         """Get the results of a program job."""
         return self._get_job(job_id).result()
+
+    def job_interim_results(self, job_id):
+        """Get the interim results of a program job."""
+        return self._get_job(job_id).interim_results()
 
     def job_cancel(self, job_id):
         """Cancel the job."""

test/ibm/runtime/test_runtime.py

@@ -676,6 +676,12 @@ def test_final_result(self):
         result = job.result()
         self.assertTrue(result)
 
+    def test_interim_results(self):
+        """Test getting interim results."""
+        job = self._run_program()
+        interim_results = job.interim_results()
+        self.assertTrue(interim_results)
+
     def test_job_status(self):
         """Test job status."""
         job = self._run_program()

test/ibm/runtime/test_runtime_integration.py

@@ -528,6 +528,14 @@ def result_callback(job_id, interim_result):
         self.assertFalse(called_back)
         self.assertIsNotNone(job._ws_client._server_close_code)
 
+    def test_retrieve_interim_results(self):
+        """Test retrieving interim results with API endpoint"""
+        int_res = "foo"
+        job = self._run_program(interim_results=int_res)
+        job.wait_for_final_state()
+        interim_results = job.interim_results()
+        self.assertIn(int_res, interim_results[0])
+
     def test_callback_error(self):
         """Test error in callback method."""
         def result_callback(job_id, interim_result):