Save local mode jobs

.gitignore

@@ -115,5 +115,8 @@ Qconfig.py
 # Generated release notes file
 docs/release_notes.rst
 
+# Locally run jobs 
+qiskit_ibm_runtime/fake_provider/local_jobs
+
 # Version.txt
 qiskit_ibm_runtime/VERSION.txt

qiskit_ibm_runtime/fake_provider/local_service.py

@@ -14,10 +14,12 @@
 
 from __future__ import annotations
 
+import os
 import math
 import copy
 import logging
 import warnings
+import pickle
 from dataclasses import asdict
 from typing import Callable, Dict, List, Literal, Optional, Union
 
@@ -49,6 +51,11 @@ def __init__(self) -> None:
             An instance of QiskitRuntimeService.
 
         """
+        self._channel_strategy = None
+        self._saved_jobs_directory = (
+            os.getenv("QISKIT_LOCAL_JOBS_DIRECTORY")
+            or f"{os.path.dirname(os.path.realpath(__file__))}/local_jobs"
+        )
 
     def backend(
         self, name: str = None, instance: str = None  # pylint: disable=unused-argument
@@ -188,12 +195,14 @@ def _run(
         inputs = copy.deepcopy(inputs)
 
         primitive_inputs = {"pubs": inputs.pop("pubs")}
-        return self._run_backend_primitive_v2(
+        job = self._run_backend_primitive_v2(
             backend=backend,
             primitive=program_id,
             options=inputs.get("options", {}),
             inputs=primitive_inputs,
         )
+        self._save_job(job)
+        return job
 
     def _run_backend_primitive_v2(
         self,
@@ -268,3 +277,40 @@ def _run_backend_primitive_v2(
             warnings.warn(f"Options {options_copy} have no effect in local testing mode.")
 
         return primitive_inst.run(**inputs)
+
+    def job(self, job_id: str) -> PrimitiveJob:
+        """Return saved local job."""
+        os.makedirs(f"{self._saved_jobs_directory}", exist_ok=True)
+        with open(f"{self._saved_jobs_directory}/{job_id}.pkl", "rb") as file:
+            return pickle.load(file)
+
+    def jobs(self) -> List[PrimitiveJob]:
+        """Return all saved local jobs."""
+        all_jobs = []
+        os.makedirs(f"{self._saved_jobs_directory}", exist_ok=True)
+        for filename in os.listdir(self._saved_jobs_directory):
+            with open(f"{self._saved_jobs_directory}/{filename}", "rb") as file:
+                all_jobs.append(pickle.load(file))
+        return all_jobs
+
+    def delete_job(self, job_id: str) -> None:
+        """Delete a local job."""
+        try:
+            os.makedirs(f"{self._saved_jobs_directory}", exist_ok=True)
+            os.remove(f"{self._saved_jobs_directory}/{job_id}.pkl")
+        except Exception as ex:  # pylint: disable=broad-except
+            logger.warning("Unable to delete job %s. %s", job_id, ex)
+
+    def _save_job(self, job: PrimitiveJob) -> None:
+        """Pickle and save job locally in the specified directory.
+
+        Args:
+            job: PrimitiveJob.
+        """
+        try:
+            job._prepare_dump()
+            os.makedirs(f"{self._saved_jobs_directory}", exist_ok=True)
+            with open(f"{self._saved_jobs_directory}/{job.job_id()}.pkl", "wb") as file:
+                pickle.dump(job, file)
+        except Exception as ex:  # pylint: disable=broad-except
+            logger.warning("Unable to save job %s. %s", job.job_id(), ex)

qiskit_ibm_runtime/qiskit_runtime_service.py

@@ -59,7 +59,7 @@ def __new__(cls, *args, **kwargs):  # type: ignore[no-untyped-def]
             # pylint: disable=import-outside-toplevel
             from .fake_provider.local_service import QiskitRuntimeLocalService
 
-            return super().__new__(QiskitRuntimeLocalService)
+            return QiskitRuntimeLocalService()
         else:
             return super().__new__(cls)
 

release-notes/unreleased/1854.feat.rst

@@ -0,0 +1,5 @@
+The methods ``job()``, ``jobs()``, and ``delete_jobs()`` have been added to 
+:class:`QiskitRuntimeLocalService`. Jobs run in local mode will now be 
+automatically saved into a local directory. This directory defaults to 
+``qiskit_ibm_runtime/fake_provider/local_jobs`` but can be customized with the 
+``QISKIT_LOCAL_JOBS_DIRECTORY`` environment variable. 

test/unit/test_local_mode.py

@@ -25,6 +25,7 @@
 from qiskit.primitives.containers.data_bin import DataBin
 
 from qiskit_ibm_runtime.fake_provider import FakeManilaV2
+from qiskit_ibm_runtime.fake_provider.local_service import QiskitRuntimeLocalService
 from qiskit_ibm_runtime import (
     Session,
     Batch,
@@ -43,6 +44,10 @@
 class TestLocalModeV2(IBMTestCase):
     """Class for testing local mode for V2 primitives."""
 
+    def setUp(self) -> None:
+        super().setUp()
+        self._service = QiskitRuntimeLocalService()
+
     @combine(backend=[FakeManilaV2(), AerSimulator()], num_sets=[1, 3])
     def test_v2_sampler(self, backend, num_sets):
         """Test V2 Sampler on a local backend."""
@@ -55,6 +60,7 @@ def test_v2_sampler(self, backend, num_sets):
             self.assertIsInstance(pub_result, SamplerPubResult)
             self.assertIsInstance(pub_result.data, DataBin)
             self.assertIsInstance(pub_result.metadata, dict)
+        self._service.delete_job(job.job_id())
 
     @combine(backend=[FakeManilaV2(), AerSimulator()], num_sets=[1, 3])
     def test_v2_estimator(self, backend, num_sets):
@@ -68,6 +74,7 @@ def test_v2_estimator(self, backend, num_sets):
             self.assertIsInstance(pub_result, PubResult)
             self.assertIsInstance(pub_result.data, DataBin)
             self.assertIsInstance(pub_result.metadata, dict)
+        self._service.delete_job(job.job_id())
 
     @data(FakeManilaV2(), AerSimulator.from_backend(FakeManilaV2()))
     def test_v2_sampler_with_accepted_options(self, backend):
@@ -77,6 +84,7 @@ def test_v2_sampler_with_accepted_options(self, backend):
         job = inst.run(**get_primitive_inputs(inst, backend=backend))
         pub_result = job.result()[0]
         self.assertEqual(pub_result.data.meas.num_shots, 10)
+        self._service.delete_job(job.job_id())
 
     @data(FakeManilaV2(), AerSimulator.from_backend(FakeManilaV2()))
     def test_v2_estimator_with_accepted_options(self, backend):
@@ -87,6 +95,7 @@ def test_v2_estimator_with_accepted_options(self, backend):
         pub_result = job.result()[0]
         self.assertIn(("target_precision", 0.03125), pub_result.metadata.items())
         self.assertTrue(pub_result.data)
+        self._service.delete_job(job.job_id())
 
     @data(FakeManilaV2(), AerSimulator.from_backend(FakeManilaV2()))
     def test_v2_estimator_with_default_shots_option(self, backend):
@@ -96,6 +105,7 @@ def test_v2_estimator_with_default_shots_option(self, backend):
         job = inst.run(**get_primitive_inputs(inst, backend=backend))
         pub_result = job.result()[0]
         self.assertIn(("target_precision", 0.1), pub_result.metadata.items())
+        self._service.delete_job(job.job_id())
 
     @combine(primitive=[SamplerV2, EstimatorV2], backend=[FakeManilaV2(), AerSimulator()])
     def test_primitive_v2_with_not_accepted_options(self, primitive, backend):
@@ -111,6 +121,7 @@ def test_primitive_v2_with_not_accepted_options(self, primitive, backend):
             _ = job.result()
             warning_messages = "".join([str(warn.message) for warn in warns])
             self.assertIn("dynamical_decoupling", warning_messages)
+        self._service.delete_job(job.job_id())
 
     @combine(session_cls=[Session, Batch], backend=[FakeManilaV2(), AerSimulator()])
     def test_sampler_v2_session(self, session_cls, backend):
@@ -125,6 +136,7 @@ def test_sampler_v2_session(self, session_cls, backend):
                 self.assertIsInstance(pub_result, PubResult)
                 self.assertIsInstance(pub_result.data, DataBin)
                 self.assertIsInstance(pub_result.metadata, dict)
+        self._service.delete_job(job.job_id())
 
     @combine(session_cls=[Session, Batch], backend=[FakeManilaV2(), AerSimulator()])
     def test_sampler_v2_session_no_params(self, session_cls, backend):
@@ -139,6 +151,7 @@ def test_sampler_v2_session_no_params(self, session_cls, backend):
                 self.assertIsInstance(pub_result, PubResult)
                 self.assertIsInstance(pub_result.data, DataBin)
                 self.assertIsInstance(pub_result.metadata, dict)
+        self._service.delete_job(job.job_id())
 
     @combine(session_cls=[Session, Batch], backend=[FakeManilaV2(), AerSimulator()])
     def test_estimator_v2_session(self, session_cls, backend):
@@ -153,10 +166,40 @@ def test_estimator_v2_session(self, session_cls, backend):
                 self.assertIsInstance(pub_result, PubResult)
                 self.assertIsInstance(pub_result.data, DataBin)
                 self.assertIsInstance(pub_result.metadata, dict)
+        self._service.delete_job(job.job_id())
 
     @data(FakeManilaV2(), AerSimulator())
     def test_non_primitive(self, backend):
         """Test calling non-primitive in local mode."""
         session = Session(backend=backend)
         with self.assertRaisesRegex(ValueError, "Only sampler and estimator"):
             session._run(program_id="foo", inputs={})
+
+    @combine(backend=[FakeManilaV2()])
+    def test_retrieve_job(self, backend):
+        """Test V2 Sampler on a local backend."""
+        inst = SamplerV2(mode=backend)
+        job = inst.run(**get_primitive_inputs(inst, backend=backend))
+        job.result()
+        rjob = self._service.job(job.job_id())
+        self.assertEqual(rjob.job_id(), job.job_id())
+        self._service.delete_job(job.job_id())
+
+    @combine(backend=[FakeManilaV2()])
+    def test_retrieve_jobs(self, backend):
+        """Test V2 Sampler on a local backend."""
+        inst = SamplerV2(mode=backend)
+        job = inst.run(**get_primitive_inputs(inst, backend=backend))
+        job.result()
+        rjobs = self._service.jobs()
+        self.assertIn(job.job_id(), [rjob.job_id() for rjob in rjobs])
+        self._service.delete_job(job.job_id())
+
+    @combine(backend=[FakeManilaV2()])
+    def test_delete_job(self, backend):
+        """Test V2 Sampler on a local backend."""
+        inst = SamplerV2(mode=backend)
+        job = inst.run(**get_primitive_inputs(inst, backend=backend))
+        job.result()
+        self._service.delete_job(job.job_id())
+        self.assertNotIn(job.job_id(), [rjob.job_id() for rjob in self._service.jobs()])