# This code is part of Qiskit.
#
# (C) Copyright IBM 2021.
#
# This code is licensed under the Apache License, Version 2.0. You may
# obtain a copy of this license in the LICENSE.txt file in the root directory
# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
#
# Any modifications or derivative works of this code must retain this
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.
"""Backend namespace for an IBM Quantum account."""
import logging
from datetime import datetime
from typing import Dict, List, Callable, Optional, Any, Union
from typing_extensions import Literal
from qiskit.providers.exceptions import QiskitBackendNotFoundError
from qiskit.providers.jobstatus import JobStatus
from qiskit.providers.providerutils import filter_backends
from qiskit.providers.models import (
PulseBackendConfiguration,
QasmBackendConfiguration,
)
# pylint: disable=unused-import
from qiskit_ibm_provider import (
ibm_backend,
ibm_provider,
)
from .api.exceptions import ApiError
from .api.clients import AccountClient
from .apiconstants import ApiJobStatus
from .exceptions import (
IBMBackendValueError,
IBMBackendApiError,
IBMBackendApiProtocolError,
IBMInputValueError,
)
from .hub_group_project import HubGroupProject
from .ibm_backend import IBMBackend, IBMRetiredBackend
from .job import IBMJob, IBMCircuitJob
from .job.exceptions import IBMJobNotFoundError
from .utils.hgp import from_instance_format
from .utils.backend_decoder import configuration_from_server_data
from .utils.converters import local_to_utc
from .utils.utils import (
to_python_identifier,
validate_job_tags,
filter_data,
api_status_to_job_status,
)
from .utils.hgp import to_instance_format
logger = logging.getLogger(__name__)
PAGE_SIZE = 50
[docs]class IBMBackendService:
"""Backend namespace for an IBM Quantum account.
Represent a namespace that provides backend related services for the IBM
Quantum backends available to this account. An instance of
this class is used as a callable attribute to the :class:`IBMProvider`
class. This allows a convenient way to query for all backends or to access
a specific backend::
backends = provider.backends() # Invoke backends() to get the backends.
sim_backend = provider.backend.ibmq_qasm_simulator # Get a specific backend instance.
Also, you are able to retrieve jobs from an account without specifying the backend name.
For example, to retrieve the ten most recent jobs you have submitted, regardless of the
backend they were submitted to, you could do::
most_recent_jobs = provider.backend.jobs(limit=10)
It is also possible to retrieve a single job without specifying the backend name::
job = provider.backend.retrieve_job(<JOB_ID>)
"""
def __init__(
self, provider: "ibm_provider.IBMProvider", hgp: HubGroupProject
) -> None:
"""IBMBackendService constructor.
Args:
provider: IBM Quantum account provider.
hgp: default hub/group/project to use for the service.
"""
super().__init__()
self._provider = provider
self._default_hgp = hgp
self._backends: Dict[str, IBMBackend] = {}
self._backend_configs: Dict[str, Any] = {}
self._initialize_backends()
def _initialize_backends(self) -> None:
"""Initialize the internal list of backends."""
# Add backends from user selected hgp followed by backends
# from other hgps if not already added
for hgp in self._provider._get_hgps():
for name in hgp.backends:
if name not in self._backends:
self._backends[name] = None
def _discover_backends(self) -> None:
"""Discovers the remote backends for this account, if not already known."""
for backend in self._backends.values():
backend_name = to_python_identifier(backend.name)
# Append _ if duplicate
while backend_name in self.__dict__:
backend_name += "_"
setattr(self, backend_name, backend)
[docs] def backends(
self,
name: Optional[str] = None,
filters: Optional[Callable[[List[IBMBackend]], bool]] = None,
min_num_qubits: Optional[int] = None,
instance: Optional[str] = None,
dynamic_circuits: Optional[bool] = None,
**kwargs: Any,
) -> List[IBMBackend]:
"""Return all backends accessible via this account, subject to optional filtering.
Args:
name: Backend name to filter by.
min_num_qubits: Minimum number of qubits the backend must have.
instance: The provider in the hub/group/project format.
dynamic_circuits: Filter by whether the backend supports dynamic circuits.
filters: More complex filters, such as lambda functions.
For example::
IBMProvider.backends(filters=lambda b: b.max_shots > 50000)
IBMProvider.backends(filters=lambda x: ("rz" in x.basis_gates )
**kwargs: Simple filters that require a specific value for an attribute in
backend configuration, backend status, or provider credentials.
Examples::
# Get the operational real backends
IBMProvider.backends(simulator=False, operational=True)
# Get the backends with at least 127 qubits
IBMProvider.backends(min_num_qubits=127)
# Get the backends that support OpenPulse
IBMProvider.backends(open_pulse=True)
For the full list of backend attributes, see the `IBMBackend` class documentation
<https://qiskit.org/documentation/apidoc/providers_models.html>
Returns:
The list of available backends that match the filter.
Raises:
IBMBackendValueError: If only one or two parameters from `hub`, `group`,
`project` are specified.
QiskitBackendNotFoundError: If the backend is not found in any instance.
"""
backends: List[IBMBackend] = []
if name:
if name not in self._backends:
raise QiskitBackendNotFoundError("No backend matches the criteria")
if not self._backends[name] or instance != self._backends[name]._instance:
self._set_backend_config(name)
self._backends[name] = self._create_backend_obj(
self._backend_configs[name], instance, self._provider._get_hgps()
)
if self._backends[name]:
backends.append(self._backends[name])
elif instance:
hgp = self._provider._get_hgp(instance=instance)
for backend_name in hgp.backends.keys():
if (
not self._backends[backend_name]
or instance != self._backends[backend_name]._instance
):
self._set_backend_config(backend_name, instance)
self._backends[backend_name] = self._create_backend_obj(
self._backend_configs[backend_name], instance
)
if self._backends[backend_name]:
backends.append(self._backends[backend_name])
else:
hgps = self._provider._get_hgps()
for backend_name, backend_config in self._backends.items():
if not backend_config:
self._set_backend_config(backend_name)
self._backends[backend_name] = self._create_backend_obj(
self._backend_configs[backend_name], hgps=hgps
)
if self._backends[backend_name]:
backends.append(self._backends[backend_name])
# Special handling of the `name` parameter, to support alias resolution.
if name:
aliases = self._aliased_backend_names()
aliases.update(self._deprecated_backend_names())
name = aliases.get(name, name)
kwargs["backend_name"] = name
if min_num_qubits:
backends = list(
filter(lambda b: b.configuration().n_qubits >= min_num_qubits, backends)
)
if dynamic_circuits is not None:
backends = list(
filter(
lambda b: (
"qasm3" in getattr(b.configuration(), "supported_features", [])
)
== dynamic_circuits,
backends,
)
)
return filter_backends(backends, filters=filters, **kwargs)
[docs] def jobs(
self,
limit: Optional[int] = 10,
skip: int = 0,
backend_name: Optional[str] = None,
status: Optional[
Union[
Literal["pending", "completed"],
List[Union[JobStatus, str]],
JobStatus,
str,
]
] = None,
start_datetime: Optional[datetime] = None,
end_datetime: Optional[datetime] = None,
job_tags: Optional[List[str]] = None,
descending: bool = True,
instance: Optional[str] = None,
legacy: bool = False,
) -> List[IBMJob]:
"""Return a list of jobs, subject to optional filtering.
Retrieve jobs that match the given filters and paginate the results
if desired. Note that the server has a limit for the number of jobs
returned in a single call. As a result, this function might involve
making several calls to the server.
Args:
limit: Number of jobs to retrieve. ``None`` means no limit. Note that the
number of sub-jobs within a composite job count towards the limit.
skip: Starting index for the job retrieval.
backend_name: Name of the backend to retrieve jobs from.
status: Filter jobs with either "pending" or "completed" status. You can also specify by
exact status. For example, `status=JobStatus.RUNNING` or `status="RUNNING"`
or `status=["RUNNING", "ERROR"]`.
start_datetime: Filter by the given start date, in local time. This is used to
find jobs whose creation dates are after (greater than or equal to) this
local date/time.
end_datetime: Filter by the given end date, in local time. This is used to
find jobs whose creation dates are before (less than or equal to) this
local date/time.
job_tags: Filter by tags assigned to jobs. Matched jobs are associated with all tags.
descending: If ``True``, return the jobs in descending order of the job
creation date (i.e. newest first) until the limit is reached.
instance: The provider in the hub/group/project format.
legacy: If ``True``, only retrieve jobs run from the archived ``qiskit-ibmq-provider``.
Otherwise, only retrieve jobs run from ``qiskit-ibm-provider``.
Returns:
A list of ``IBMJob`` instances.
Raises:
IBMBackendValueError: If a keyword value is not recognized.
TypeError: If the input `start_datetime` or `end_datetime` parameter value
is not valid.
"""
# Build the filter for the query.
api_filter = {} # type: Dict[str, Any]
all_job_statuses = [status.name for status in JobStatus]
if isinstance(status, JobStatus):
status = status.name
if isinstance(status, str):
status = status.upper()
if isinstance(status, list):
status = [x.name if isinstance(x, JobStatus) else x.upper() for x in status]
if status in (["INITIALIZING"], ["VALIDATING"]):
return []
elif all(x in ["DONE", "CANCELLED", "ERROR"] for x in status):
api_filter["pending"] = False
elif all(x in ["QUEUED", "RUNNING"] for x in status):
api_filter["pending"] = True
elif status in all_job_statuses:
if status in ["INITIALIZING", "VALIDATING"]:
return []
elif status in ["DONE", "CANCELLED", "ERROR"]:
api_filter["pending"] = False
elif status in ["QUEUED", "RUNNING"]:
api_filter["pending"] = True
if backend_name:
api_filter["backend"] = backend_name
if status == "PENDING":
api_filter["pending"] = True
if status == "COMPLETED":
api_filter["pending"] = False
if start_datetime:
api_filter["created_after"] = local_to_utc(start_datetime).isoformat()
if end_datetime:
api_filter["created_before"] = local_to_utc(end_datetime).isoformat()
if job_tags:
validate_job_tags(job_tags, IBMBackendValueError)
api_filter["job_tags"] = job_tags
if instance:
hub, group, project = from_instance_format(instance)
api_filter["hub"] = hub
api_filter["group"] = group
api_filter["project"] = project
# Retrieve all requested jobs.
filter_by_status = (
status
and status not in ["PENDING", "COMPLETED"]
and (
status in all_job_statuses or all(x in all_job_statuses for x in status)
)
)
job_list = []
while True:
job_responses = self._get_jobs(
api_filter=api_filter,
limit=limit,
skip=skip,
descending=descending,
legacy=legacy,
)
if len(job_responses) == 0:
break
for job_info in job_responses:
if (
job_info.get("program", {}).get("id")
in [
"circuit-runner",
"qasm3-runner",
]
or legacy
):
if filter_by_status:
if legacy:
job_info_status = job_info["status"].upper()
else:
job_info_status = job_info["state"]["status"].upper()
job_status = api_status_to_job_status(job_info_status).name
if (isinstance(status, str) and job_status != status) or (
isinstance(status, list) and job_status not in status
):
continue
job = self._restore_circuit_job(
job_info, raise_error=False, legacy=legacy
)
if job is None:
logger.warning(
'Discarding job "%s" because it contains invalid data.',
job_info.get("job_id", ""),
)
continue
job_list.append(job)
if limit and len(job_list) == limit:
return job_list
skip += len(job_responses)
return job_list
def _get_jobs(
self,
api_filter: Dict,
limit: Optional[int] = 10,
skip: int = 0,
descending: bool = True,
legacy: bool = False,
) -> List:
"""Retrieve the requested number of jobs from the server using pagination.
Args:
api_filter: Filter used for querying.
limit: Number of jobs to retrieve. ``None`` means no limit.
skip: Starting index for the job retrieval.
descending: If ``True``, return the jobs in descending order of the job
creation date (i.e. newest first) until the limit is reached.
legacy: Filter to only retrieve jobs run from the archived ``qiskit-ibmq-provider``.
Returns:
A list of raw API response.
"""
# Retrieve the requested number of jobs, using pagination. The server
# might limit the number of jobs per request.
job_responses: List[Dict[str, Any]] = []
current_page_limit = (
limit if (limit is not None and limit <= PAGE_SIZE) else PAGE_SIZE
)
while True:
if legacy:
job_page = self._default_hgp._api_client.list_jobs(
limit=current_page_limit,
skip=skip,
descending=descending,
extra_filter=api_filter,
)
else:
job_page = self._provider._runtime_client.jobs_get(
limit=current_page_limit,
skip=skip,
descending=descending,
**api_filter,
)["jobs"]
if logger.getEffectiveLevel() is logging.DEBUG:
filtered_data = [filter_data(job) for job in job_page]
logger.debug("jobs() response data is %s", filtered_data)
if not job_page:
# Stop if there are no more jobs returned by the server.
break
job_responses += job_page
if limit:
if len(job_responses) >= limit:
# Stop if we have reached the limit.
break
current_page_limit = limit - len(job_responses)
else:
current_page_limit = PAGE_SIZE
skip = len(job_responses)
return job_responses
def _restore_circuit_job(
self, job_info: Dict, raise_error: bool, legacy: bool = False
) -> Optional[IBMCircuitJob]:
"""Restore a circuit job from the API response.
Args:
job_info: Job info in dictionary format.
raise_error: Whether to raise an exception if `job_info` is in
an invalid format.
legacy: Filter to only retrieve jobs run from the archived ``qiskit-ibmq-provider``.
Returns:
Circuit job restored from the data, or ``None`` if format is invalid.
Raises:
IBMBackendApiProtocolError: If unexpected return value received
from the server.
"""
if legacy:
job_id = job_info.get("job_id", "")
backend_name = job_info.get("_backend_info", {}).get("name", "unknown")
instance = None
job_params = job_info
else:
job_id = job_info["id"]
job_params = {
"job_id": job_id,
"creation_date": job_info["created"],
"status": job_info["status"],
"runtime_client": self._provider._runtime_client,
"tags": job_info.get("tags"),
}
# Recreate the backend used for this job.
backend_name = job_info.get("backend")
instance = to_instance_format(
job_info["hub"], job_info["group"], job_info["project"]
)
try:
backend = self._provider.get_backend(backend_name, instance)
except QiskitBackendNotFoundError:
backend = IBMRetiredBackend.from_name(
backend_name=backend_name,
provider=self._provider,
api=self._default_hgp._api_client,
)
try:
job = IBMCircuitJob(
backend=backend, api_client=self._default_hgp._api_client, **job_params
)
return job
except TypeError as ex:
if raise_error:
raise IBMBackendApiProtocolError(
f"Unexpected return value received from the server "
f"when retrieving job {job_id}: {ex}"
) from ex
return None
def _merge_logical_filters(self, cur_filter: Dict, new_filter: Dict) -> None:
"""Merge the logical operators in the input filters.
Args:
cur_filter: Current filter.
new_filter: New filter to be merged into ``cur_filter``.
Returns:
``cur_filter`` with ``new_filter``'s logical operators merged into it.
"""
logical_operators_to_expand = ["or", "and"]
for key in logical_operators_to_expand:
if key in new_filter:
if key in cur_filter:
cur_filter[key].extend(new_filter[key])
else:
cur_filter[key] = new_filter[key]
def _update_creation_date_filter(
self,
cur_dt_filter: Dict[str, Any],
gte_dt: Optional[str] = None,
lte_dt: Optional[str] = None,
) -> Dict[str, Any]:
"""Use the new start and end datetime in the creation date filter.
Args:
cur_dt_filter: Current creation date filter.
gte_dt: New start datetime.
lte_dt: New end datetime.
Returns:
Updated creation date filter.
"""
if not gte_dt:
gt_list = [
cur_dt_filter.pop(gt_op)
for gt_op in ["gt", "gte"]
if gt_op in cur_dt_filter
]
if "between" in cur_dt_filter and len(cur_dt_filter["between"]) > 0:
gt_list.append(cur_dt_filter.pop("between")[0])
gte_dt = max(gt_list) if gt_list else None
if not lte_dt:
lt_list = [
cur_dt_filter.pop(lt_op)
for lt_op in ["lt", "lte"]
if lt_op in cur_dt_filter
]
if "between" in cur_dt_filter and len(cur_dt_filter["between"]) > 1:
lt_list.append(cur_dt_filter.pop("between")[1])
lte_dt = min(lt_list) if lt_list else None
new_dt_filter = {} # type: Dict[str, Union[str, List[str]]]
if gte_dt and lte_dt:
new_dt_filter["between"] = [gte_dt, lte_dt]
elif gte_dt:
new_dt_filter["gte"] = gte_dt
elif lte_dt:
new_dt_filter["lte"] = lte_dt
return new_dt_filter
def _get_status_db_filter(
self, status_arg: Union[JobStatus, str, List[Union[JobStatus, str]]]
) -> Dict[str, Any]:
"""Return the db filter to use when retrieving jobs based on a status or statuses.
Returns:
The status db filter used to query the api when retrieving jobs that match
a given status or list of statuses.
Raises:
IBMBackendError: If a status value is not recognized.
"""
_final_status_filter = None
if isinstance(status_arg, list):
_final_status_filter = {"or": []}
for status in status_arg:
status_filter = self._get_status_filter(status)
_final_status_filter["or"].append(status_filter)
else:
status_filter = self._get_status_filter(status_arg)
_final_status_filter = status_filter
return _final_status_filter
def _get_status_filter(self, status: Union[JobStatus, str]) -> Dict[str, Any]:
"""Return the db filter to use when retrieving jobs based on a status.
Returns:
The status db filter used to query the api when retrieving jobs
that match a given status.
Raises:
IBMBackendValueError: If the status value is not recognized.
"""
if isinstance(status, str):
try:
status = JobStatus[status.upper()]
except KeyError:
raise IBMBackendValueError(
'"{}" is not a valid status value. Valid values are {}'.format(
status, ", ".join(job_status.name for job_status in JobStatus)
)
) from None
_status_filter = {} # type: Dict[str, Any]
if status == JobStatus.INITIALIZING:
_status_filter = {
"status": {
"inq": [ApiJobStatus.CREATING.value, ApiJobStatus.CREATED.value]
}
}
elif status == JobStatus.VALIDATING:
_status_filter = {
"status": {
"inq": [ApiJobStatus.VALIDATING.value, ApiJobStatus.VALIDATED.value]
}
}
elif status == JobStatus.RUNNING:
_status_filter = {"status": ApiJobStatus.RUNNING.value}
elif status == JobStatus.QUEUED:
_status_filter = {"status": ApiJobStatus.QUEUED.value}
elif status == JobStatus.CANCELLED:
_status_filter = {"status": ApiJobStatus.CANCELLED.value}
elif status == JobStatus.DONE:
_status_filter = {"status": ApiJobStatus.COMPLETED.value}
elif status == JobStatus.ERROR:
_status_filter = {"status": {"regexp": "^ERROR"}}
else:
raise IBMBackendValueError(
'"{}" is not a valid status value. Valid values are {}'.format(
status, ", ".join(job_status.name for job_status in JobStatus)
)
)
return _status_filter
[docs] def retrieve_job(self, job_id: str) -> IBMJob:
"""Return a single job.
Args:
job_id: The ID of the job to retrieve.
Returns:
The job with the given id.
Raises:
IBMBackendApiError: If an unexpected error occurred when retrieving
the job.
IBMBackendApiProtocolError: If unexpected return value received
from the server.
IBMJobNotFoundError: If job cannot be found.
IBMInputValueError: If job exists but was run from a different service.
"""
try:
legacy = False
if self._provider._runtime_client.job_type(job_id) == "IQX":
legacy = True
job_info = self._default_hgp._api_client.job_get(job_id)
else:
job_info = self._provider._runtime_client.job_get(
job_id, exclude_params=True
)
if job_info.get("program", {}).get("id") not in [
"circuit-runner",
"qasm3-runner",
]:
raise IBMInputValueError(
f"Job {job_id} was not run with qiskit-ibm-provider, "
f"please try retrieving job results from qiskit-ibm-runtime"
)
except ApiError as ex:
if "Error code: 3250." in str(ex):
raise IBMJobNotFoundError(f"Job {job_id} not found.")
raise IBMBackendApiError(
"Failed to get job {}: {}".format(job_id, str(ex))
) from ex
job = self._restore_circuit_job(job_info, raise_error=True, legacy=legacy)
return job
def _set_backend_config(
self, backend_name: str, instance: Optional[str] = None
) -> None:
"""Retrieve backend configuration and add to backend_configs.
Args:
backend_name: backend name that will be returned.
instance: the current h/g/p.
"""
if backend_name not in self._backend_configs:
raw_config = self._provider._runtime_client.backend_configuration(
backend_name
)
config = configuration_from_server_data(
raw_config=raw_config, instance=instance
)
self._backend_configs[backend_name] = config
def _create_backend_obj(
self,
config: Union[QasmBackendConfiguration, PulseBackendConfiguration],
instance: Optional[str] = None,
hgps: Optional[List] = None,
) -> IBMBackend:
"""Given a backend configuration return the backend object.
Args:
config: backend configuration.
instance: the current h/g/p.
Returns:
A backend object.
Raises:
QiskitBackendNotFoundError: if the backend is not in the hgp passed in.
"""
if config:
if not instance:
for hgp in hgps:
if config.backend_name in hgp.backends:
instance = to_instance_format(
hgp._hub, hgp._group, hgp._project
)
break
elif (
config.backend_name
not in self._provider._get_hgp(instance=instance).backends
):
raise QiskitBackendNotFoundError(
f"Backend {config.backend_name} is not in "
f"{instance}: please try a different hub/group/project."
)
return ibm_backend.IBMBackend(
instance=instance,
configuration=config,
api_client=AccountClient(self._provider._client_params),
provider=self._provider,
)
return None
@staticmethod
def _deprecated_backend_names() -> Dict[str, str]:
"""Returns deprecated backend names."""
return {
"ibmqx_qasm_simulator": "ibmq_qasm_simulator",
"ibmqx_hpc_qasm_simulator": "ibmq_qasm_simulator",
"real": "ibmqx1",
}
@staticmethod
def _aliased_backend_names() -> Dict[str, str]:
"""Returns aliased backend names."""
return {
"ibmq_5_yorktown": "ibmqx2",
"ibmq_5_tenerife": "ibmqx4",
"ibmq_16_rueschlikon": "ibmqx5",
"ibmq_20_austin": "QS1_1",
}