Allow running a program without backend on cloud

qiskit_ibm_runtime/api/clients/runtime.py

@@ -116,7 +116,7 @@ def set_program_visibility(self, program_id: str, public: bool) -> None:
     def program_run(
         self,
         program_id: str,
-        backend_name: str,
+        backend_name: Optional[str],
         params: Dict,
         image: str,
         hgp: Optional[str],

qiskit_ibm_runtime/api/rest/runtime.py

@@ -113,7 +113,7 @@ def create_program(
     def program_run(
         self,
         program_id: str,
-        backend_name: str,
+        backend_name: Optional[str],
         params: Dict,
         image: str,
         hub: Optional[str] = None,
@@ -137,10 +137,11 @@ def program_run(
         url = self.get_url("jobs")
         payload = {
             "program_id": program_id,
-            "backend": backend_name,
             "params": params,
             "runtime": image,
         }
+        if backend_name:
+            payload["backend"] = backend_name
         if all([hub, group, project]):
             payload["hub"] = hub
             payload["group"] = group

qiskit_ibm_runtime/ibm_runtime_service.py

@@ -761,8 +761,8 @@ def _to_program(self, response: Dict) -> RuntimeProgram:
     def run(
         self,
         program_id: str,
-        options: Dict,
         inputs: Union[Dict, ParameterNamespace],
+        options: Optional[Dict] = None,
         callback: Optional[Callable] = None,
         result_decoder: Optional[Type[ResultDecoder]] = None,
         image: str = "",
@@ -772,10 +772,11 @@ def run(
 
         Args:
             program_id: Program ID.
-            options: Runtime options that control the execution environment.
-                Currently the only available option is ``backend_name``, which is required.
             inputs: Program input parameters. These input values are passed
                 to the runtime program.
+            options: Runtime options that control the execution environment.
+                Currently the only available option is ``backend_name``, which is required if
+                you are using legacy runtime.
             callback: Callback function to be invoked for any interim results.
                 The callback function will receive 2 positional parameters:
 
@@ -808,7 +809,9 @@ def run(
         ):
             raise IBMInputValueError('"image" needs to be in form of image_name:tag')
 
-        backend_name = options.get("backend_name", "")
+        options = options or {}
+        backend_name = options.get("backend_name", None)
+        backend = None
 
         hgp_name = None
         if self._auth == "legacy":
@@ -820,10 +823,6 @@ def run(
             hgp = self._get_hgp(instance=instance, backend_name=backend_name)
             backend = hgp.backend(backend_name)
             hgp_name = hgp.name
-        else:
-            # TODO Support instance for cloud
-            # TODO Support optional backend name when fully supported by server
-            backend = self.backend(backend_name)
 
         result_decoder = result_decoder or ResultDecoder
         try:
@@ -841,6 +840,9 @@ def run(
                 ) from None
             raise IBMRuntimeError(f"Failed to run program: {ex}") from None
 
+        if not backend:
+            backend = self.backend(name=response["backend"])
+
         job = RuntimeJob(
             backend=backend,
             api_client=self._api_client,

test/ibm_test_case.py

@@ -187,7 +187,9 @@ def _run_program(
             }
         )
         pid = program_id or self.program_ids[service.auth]
-        backend_name = backend or self.sim_backends[service.auth]
+        backend_name = (
+            backend if backend is not None else self.sim_backends[service.auth]
+        )
         options = {"backend_name": backend_name}
         job = service.run(
             program_id=pid, inputs=inputs, options=options, callback=callback

test/mock/fake_runtime_client.py

@@ -341,7 +341,7 @@ def program_get(self, program_id: str):
     def program_run(
         self,
         program_id: str,
-        backend_name: str,
+        backend_name: Optional[str],
         params: Dict,
         image: str,
         hgp: Optional[str],
@@ -358,6 +358,10 @@ def program_run(
             hub, group, project = from_instance_format(hgp)
         else:
             hub = group = project = None
+
+        if backend_name is None:
+            backend_name = self.list_backends()[0]
+
         job = job_cls(
             job_id=job_id,
             program_id=program_id,
@@ -371,7 +375,7 @@ def program_run(
             **self._job_kwargs,
         )
         self._jobs[job_id] = job
-        return {"id": job_id}
+        return {"id": job_id, "backend": backend_name}
 
     def program_delete(self, program_id: str) -> None:
         """Delete the specified program."""

test/test_integration_job.py

@@ -56,6 +56,12 @@ def test_run_program_real_device(self, service):
         self.assertEqual(JobStatus.DONE, job.status())
         self.assertEqual("foo", result)
 
+    def test_run_program_cloud_no_backend(self):
+        """Test running a cloud program with no backend."""
+        service = [serv for serv in self.services if serv.auth == "cloud"][0]
+        job = self._run_program(service, backend="")
+        self.assertTrue(job.backend, f"Job {job.job_id} has no backend.")
+
     @run_cloud_legacy_real
     def test_run_program_failed(self, service):
         """Test a failed program execution."""

test/test_jobs.py

@@ -74,6 +74,12 @@ def test_run_program_missing_backend_legacy(self):
         with self.assertRaises(IBMInputValueError):
             _ = run_program(service=service, backend_name="")
 
+    def test_run_program_missing_backend_cloud(self):
+        """Test running a cloud program with no backend."""
+        service = FakeRuntimeService(auth="cloud", token="my_token", instance="crn:123")
+        job = run_program(service=service, backend_name="")
+        self.assertTrue(job.backend)
+
     def test_run_program_default_hgp_backend(self):
         """Test running a program with a backend in default hgp."""
         service = FakeRuntimeService(auth="legacy", token="my_token")

test/utils/decorators.py

@@ -291,7 +291,6 @@ def run_cloud_legacy_real(func):
     @wraps(func)
     def _wrapper(self, *args, **kwargs):
         for service in self.services:
-            # for service, instance in [(legacy_service, legacy_instance)]:
             with self.subTest(service=service.auth):
                 kwargs["service"] = service
                 func(self, *args, **kwargs)