Retrieve interim results from runtime jobs

qiskit_ibm_runtime/api/clients/runtime.py

@@ -237,6 +237,17 @@ def job_results(self, job_id: str) -> str:
         """
         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

@@ -282,7 +282,13 @@ def update_metadata(
 class ProgramJob(RestAdapterBase):
     """Rest adapter for program job related endpoints."""
 
-    URL_MAP = {"self": "", "results": "/results", "cancel": "/cancel", "logs": "/logs"}
+    URL_MAP = {
+        "self": "",
+        "results": "/results",
+        "cancel": "/cancel",
+        "logs": "/logs",
+        "interim_results": "/interim_results",
+    }
 
     def __init__(
         self, session: RetrySession, job_id: str, url_prefix: str = ""
@@ -308,6 +314,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

@@ -109,13 +109,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]
@@ -130,6 +132,28 @@ 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

@@ -109,6 +109,7 @@ def __init__(
         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
@@ -143,6 +144,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."""
@@ -422,6 +427,10 @@ 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."""
         self._get_job(job_id).cancel()

test/ibm/runtime/test_runtime.py

@@ -728,6 +728,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

@@ -534,6 +534,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."""