Reintroduce delete_job (Qiskit#2541)

* Reintroduce delete_job

* Add release note

Author: diego-plan9

qiskit_ibm_runtime/qiskit_runtime_service.py

@@ -1156,6 +1156,25 @@ def jobs(
 
         return [self._decode_job(job) for job in job_responses]
 
+    def delete_job(self, job_id: str) -> None:
+        """Delete a runtime job.
+
+        Note that this operation cannot be reversed.
+
+        Args:
+            job_id: ID of the job to delete.
+
+        Raises:
+            RuntimeJobNotFound: The job doesn't exist.
+            IBMRuntimeError: Method is not supported.
+        """
+        try:
+            self._active_api_client.job_delete(job_id)
+        except RequestsApiError as ex:
+            if ex.status_code == 404:
+                raise RuntimeJobNotFound(f"Job not found: {ex.message}") from None
+            raise IBMRuntimeError(f"Failed to delete job: {ex}") from None
+
     def usage(self) -> dict[str, Any]:
         """Return usage information for the current active instance.
 

release-notes/unreleased/2541.feature.rst

@@ -0,0 +1 @@
+The ``delete_job()`` method has been reintroduced.

test/integration/test_job.py

@@ -15,16 +15,19 @@
 import random
 import time
 
+from qiskit.transpiler.preset_passmanagers import generate_preset_pass_manager
+
 from qiskit_ibm_runtime.exceptions import (
     RuntimeInvalidStateError,
+    RuntimeJobNotFound,
 )
 
 from ..ibm_test_case import IBMIntegrationJobTestCase
 from ..decorators import run_integration_test, production_only
 from ..serialization import (
     SerializableClass,
 )
-from ..utils import cancel_job_safe, wait_for_status
+from ..utils import bell, cancel_job_safe, get_real_device, wait_for_status
 
 
 class TestIntegrationJob(IBMIntegrationJobTestCase):
@@ -84,6 +87,35 @@ def test_cancel_job_done(self, service):
         with self.assertRaises(RuntimeInvalidStateError):
             job.cancel()
 
+    @run_integration_test
+    def test_delete_job(self, service):
+        """Test deleting a job."""
+        self.skipTest("Deleting jobs not supported on IBM Quantum Platform.")
+        sub_tests = ["DONE"]
+        for status in sub_tests:
+            with self.subTest(status=status):
+                job = self._run_program(service)
+                wait_for_status(job, status)
+                service.delete_job(job.job_id())
+                with self.assertRaises(RuntimeJobNotFound):
+                    service.job(job.job_id())
+
+    @run_integration_test
+    @production_only
+    def test_delete_job_queued(self, service):
+        """Test deleting a queued job."""
+        self.skipTest("Deleting jobs not supported on IBM Quantum Platform.")
+        real_device_name = get_real_device(service)
+        real_device = service.backend(real_device_name)
+        pm = generate_preset_pass_manager(optimization_level=1, target=real_device.target)
+        isa_circuit = pm.run([bell()])
+        _ = self._run_program(service, circuits=isa_circuit, backend=real_device_name)
+        job = self._run_program(service, circuits=isa_circuit, backend=real_device_name)
+        wait_for_status(job, "QUEUED")
+        service.delete_job(job.job_id())
+        with self.assertRaises(RuntimeJobNotFound):
+            service.job(job.job_id())
+
     @run_integration_test
     def test_job_status(self, service):
         """Test job status."""

test/unit/test_jobs.py

@@ -22,6 +22,7 @@
 from qiskit_ibm_runtime.constants import API_TO_JOB_ERROR_MESSAGE
 from qiskit_ibm_runtime.exceptions import (
     RuntimeJobFailureError,
+    RuntimeJobNotFound,
     RuntimeJobMaxTimeoutError,
     RuntimeInvalidStateError,
 )
@@ -147,6 +148,16 @@ def test_wait_for_final_state(self, service):
             job.wait_for_final_state()
         self.assertEqual("DONE", job.status())
 
+    @run_cloud_fake
+    def test_delete_job(self, service):
+        """Test deleting a job."""
+        params = {"param1": "foo"}
+        job = run_program(service=service, inputs=params)
+        self.assertTrue(job.job_id())
+        service.delete_job(job.job_id())
+        with self.assertRaises(RuntimeJobNotFound):
+            service.job(job.job_id())
+
     @run_cloud_fake
     def test_instance_limit_warning(self, service):
         """Test emitting a warning if instance usage has been reached."""