Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 35 additions & 3 deletions qiskit_ibm_runtime/qiskit_runtime_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -726,6 +726,7 @@ def _create_backend_obj(
instance: str,
use_fractional_gates: bool | None,
calibration_id: str | None = None,
cache: bool = True,

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This change allows for "whenever a backend object needs to be created, have the caller decide whether its configuration is cached in the state variables of the service".

Without this change, backends that were created in the new code in .backend() would remain tied to the instance that created them (as they would be cached to that instance). Since the new feature is meant to be used in very specific cases (ie. retrieving a backend manually) and to be superseeded soon, relinquishing caching of that backend seems like a sane trade-off (still allows for ensuring that the created backend belongs to the right instance, at the expense of some extra API calls when the same backend is requested multiple times).

) -> IBMBackend:
"""Given a backend configuration return the backend object.

Expand All @@ -738,6 +739,7 @@ def _create_backend_obj(
operations. See :meth:`~.QiskitRuntimeService.backends` for
further details.
calibration_id: The calibration id to use for the IBM backend.
cache: If ``False``, do not cache the backend in `self._backend_configs`.

Returns:
A backend object.
Expand Down Expand Up @@ -774,7 +776,8 @@ def _create_backend_obj(
instance=instance,
use_fractional_gates=use_fractional_gates,
)
self._backend_configs[backend_name] = config
if cache:
self._backend_configs[backend_name] = config

else:
config = configuration_from_server_data(
Expand All @@ -787,7 +790,8 @@ def _create_backend_obj(
# I know we have a configuration_registry in the api client
# but that doesn't work with new IQP since we different api clients are being used

self._backend_configs[backend_name] = config
if cache:
self._backend_configs[backend_name] = config
except Exception as ex:
logger.warning("Unable to create configuration for %s. %s ", backend_name, ex)
raise QiskitBackendNotFoundError(
Expand Down Expand Up @@ -964,7 +968,7 @@ def backend(
from qiskit_ibm_runtime import QiskitRuntimeService

service = QiskitRuntimeService()
backend = service.backend()
backend = service.backend("ibm_kingston")

status = backend.status()
assert status.operational and status.status_msg == "active"
Expand Down Expand Up @@ -992,6 +996,34 @@ def backend(
use_fractional_gates=use_fractional_gates,
calibration_id=calibration_id,
)

# `self.backends()` might not include all the backends by default. If no backend was
# returned, make a one-time uncached attempt to retrieve the backend based on its name.
if not backends:
# Use the specified instance crns, or traverse instances in sensible order.
instances = [data[0] for data in self._resolve_cloud_instances(instance)]

for instance_ in instances:
try:
self._get_or_create_cloud_client(instance_)
backends = [
self._create_backend_obj(
name, instance_, use_fractional_gates, calibration_id, cache=False
)
]
# Show a warning only if the instance is guessed.
if not instance and not self._instance_auto:
for inst_details in self._backend_instance_groups:
if instance_ == inst_details["crn"]:
logger.warning(
"Using instance: %s, plan: %s",
inst_details["name"],
inst_details["plan"],
)
break
except QiskitBackendNotFoundError:
pass

if not backends:
cloud_msg_url = ""
if self._channel in ["ibm_cloud", "ibm_quantum_platform"]:
Expand Down
76 changes: 75 additions & 1 deletion test/unit/test_backend_retrieval.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@

from __future__ import annotations

import json
from typing import TYPE_CHECKING
from unittest import mock

from ddt import ddt, named_data
Expand All @@ -25,7 +27,12 @@

from ..decorators import mock_responses
from ..ibm_test_case import IBMTestCase
from ..registries import Backend, OneInstanceNoBackendsRegistry
from ..registries import Backend, DefaultRegistry, OneInstanceNoBackendsRegistry

if TYPE_CHECKING:
from requests import PreparedRequest

from test.registries import CallbackResult


class TestBackendFilters(IBMTestCase):
Expand Down Expand Up @@ -222,6 +229,14 @@ def test_filter_min_num_qubits(self, registry):
self.assertGreaterEqual(backend.configuration().n_qubits, n_qubits)


class EmptyBackendListRegistry(DefaultRegistry):
"""Registry that returns an empty list for the `/backends` endpoint."""

def callback_backends(self, request: PreparedRequest) -> CallbackResult:
"""Callback for the IBM Quantum Compute API ``/backends`` endpoint."""
return (200, {"Content-Type": "application/json"}, json.dumps({"devices": []}))


@ddt
class TestGetBackend(IBMTestCase):
"""Test getting a backend."""
Expand Down Expand Up @@ -333,3 +348,62 @@ def test_backend_with_invalid_calibration(self, registry):

with self.assertRaises(QiskitBackendNotFoundError):
service.backend("ibm_torino", calibration_id="invalid")

@mock_responses(EmptyBackendListRegistry)
def test_backend_not_in_backends_list(self, registry):
"""Test retrieving a backend that is not in the list of backends.

This test exercises the case where a backend is retrieved via `backend()`, and that backend
is not returned in the `backends()` method.
"""
instance_a = registry.instances["a"]
instance_b = registry.instances["b"]

service = QiskitRuntimeService(token="my_token")
# Ensure that no backend appears in the backends list.
self.assertEqual(service.backends(), [])

# Retrieve an existing backend (available in several instances).
backend = service.backend("common_backend")
self.assertEqual(backend.name, "common_backend")
self.assertEqual(backend._instance, instance_a.crn)

# Retrieve an existing backend (available in several instances), passing instance.
with self.assertNoLogs("qiskit_ibm_runtime", level="WARNING"):
backend = service.backend("common_backend", instance="b")
self.assertEqual(backend.name, "common_backend")
self.assertEqual(backend._instance, instance_b.crn)

# Retrieve an existing backend (available in one instance).
backend = service.backend("unique_backend_a")
self.assertEqual(backend.name, "unique_backend_a")
self.assertEqual(backend._instance, instance_a.crn)

# Retrieve an existing backend (available in one instance), with wrong instance.
with (
self.assertRaises(QiskitBackendNotFoundError),
self.assertNoLogs("qiskit_ibm_runtime", level="WARNING"),
):
backend = service.backend("unique_backend_a", instance="b")

@mock_responses(EmptyBackendListRegistry)
def test_backend_not_in_backends_list_instance_auto(self, registry):
"""Test retrieving a backend not in the list of backends, with service instance `auto`.

This test exercises the case where a backend is retrieved via `backend()`, and that backend
is not returned in the `backends()` method.

When passing `instance=auto` to `QiskitRuntimeService()`, warnings should not be emitted
when guessing instances.
"""
instance_a = registry.instances["a"]

service = QiskitRuntimeService(token="my_token", instance="auto")
# Ensure that no backend appears in the backends list.
self.assertEqual(service.backends(), [])

# Retrieve an existing backend (available in one instance).
with self.assertNoLogs("qiskit_ibm_runtime", level="WARNING"):
backend = service.backend("unique_backend_a")
self.assertEqual(backend.name, "unique_backend_a")
self.assertEqual(backend._instance, instance_a.crn)
Loading