Source code for qiskit_ionq.ionq_client

# This code is part of Qiskit.
#
# (C) Copyright IBM 2017, 2018.
#
# 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.

# Copyright 2020 IonQ, Inc. (www.ionq.com)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Basic API Client for IonQ's REST API"""

from __future__ import annotations

import re
import json
from collections import OrderedDict
from datetime import datetime
from typing import Optional, TYPE_CHECKING
from warnings import warn
import requests

from . import exceptions
from .helpers import qiskit_to_ionq, get_user_agent, retry
from .exceptions import IonQRetriableError

if TYPE_CHECKING:  # pragma: no cover
    from .ionq_job import IonQJob


[docs] class IonQClient: """IonQ API Client Attributes: _url(str): A URL base to use for API calls, e.g. ``"https://api.ionq.co/v0.4"`` _token(str): An API Access Token to use with the IonQ API. _custom_headers(dict): Extra headers to add to the request. """ def __init__( self, token: Optional[str] = None, url: Optional[str] = None, custom_headers: Optional[dict] = None, ): self._token = token self._custom_headers = custom_headers or {} # strip trailing slashes from our base URL. if url and url.endswith("/"): url = url[:-1] self._url = url self._user_agent = get_user_agent() @property def api_headers(self) -> dict: """API Headers needed to make calls to the REST API. Returns: dict[str, str]: A dict of :class:`requests.Request` headers. """ return { **self._custom_headers, "Authorization": f"apiKey {self._token}", "Content-Type": "application/json", "User-Agent": self._user_agent, }
[docs] def make_path(self, *parts: str) -> str: """Make a "/"-delimited path, then append it to :attr:`_url`. Returns: str: A URL to use for an API call. """ return f"{self._url}/{'/'.join(parts)}"
[docs] def get_with_retry(self, req_path, params=None, headers=None, timeout=30): """Make a GET request with retry logic and exception handling. Args: req_path (str): The URL path to make the request to. params (dict, optional): Parameters to include in the request. headers (dict, optional): Headers to include in the request. timeout (int, optional): Timeout for the request. Raises: IonQRetriableError: When a retriable error occurs during the request. Returns: Response: A requests.Response object. """ try: res = requests.get( req_path, params=params, headers=headers, timeout=timeout, ) except requests.exceptions.RequestException as req_exc: raise IonQRetriableError(req_exc) from req_exc return res
[docs] @retry(exceptions=IonQRetriableError, tries=5) def submit_job(self, job: IonQJob) -> dict: """Submit job to IonQ API This returns a JSON dict with status "submitted" and the job's id. Args: job (IonQJob): The IonQ Job instance to submit to the API. Raises: IonQAPIError: When the API returns a non-200 status code. Returns: dict: A :mod:`requests <requests>` response :meth:`json <requests.Response.json>` dict. """ as_json = qiskit_to_ionq( job.circuit, job.backend(), job._passed_args, job.extra_query_params, job.extra_metadata, ) req_path = self.make_path("jobs") res = requests.post( req_path, data=as_json, headers=self.api_headers, timeout=30, ) exceptions.IonQAPIError.raise_for_status(res) return res.json()
[docs] @retry(exceptions=IonQRetriableError, max_delay=60, backoff=2, jitter=1) def retrieve_job(self, job_id: str) -> dict: """Retrieve job information from the IonQ API. The returned JSON dict will only have data if job has completed. Args: job_id (str): The ID of a job to retrieve. Raises: IonQAPIError: When the API returns a non-200 status code. IonQRetriableError: When a retriable error occurs during the request. Returns: dict: A :mod:`requests <requests>` response :meth:`json <requests.Response.json>` dict. """ req_path = self.make_path("jobs", job_id) res = self.get_with_retry(req_path, headers=self.api_headers) exceptions.IonQAPIError.raise_for_status(res) return res.json()
[docs] @retry(exceptions=IonQRetriableError, tries=5) def cancel_job(self, job_id: str) -> dict: """Attempt to cancel a job which has not yet run. .. NOTE:: If the job has already reached status "completed", this cancel action is a no-op. Args: job_id (str): The ID of the job to cancel. Raises: IonQAPIError: When the API returns a non-200 status code. Returns: dict: A :mod:`requests <requests>` response :meth:`json <requests.Response.json>` dict. """ req_path = self.make_path("jobs", job_id, "status", "cancel") res = requests.put(req_path, headers=self.api_headers, timeout=30) exceptions.IonQAPIError.raise_for_status(res) return res.json()
[docs] def cancel_jobs(self, job_ids: list[str]) -> list[dict]: """Cancel multiple jobs at once. Args: job_ids (list): A list of job IDs to cancel. Returns: list: A list of :meth:`cancel_job <cancel_job>` responses. """ return [self.cancel_job(job_id) for job_id in job_ids]
[docs] @retry(exceptions=IonQRetriableError, tries=3) def delete_job(self, job_id: str) -> dict: """Delete a job and associated data. Args: job_id (str): The ID of the job to delete. Raises: IonQAPIError: When the API returns a non-200 status code. Returns: dict: A :mod:`requests <requests>` response :meth:`json <requests.Response.json>` dict. """ req_path = self.make_path("jobs", job_id) res = requests.delete(req_path, headers=self.api_headers, timeout=30) exceptions.IonQAPIError.raise_for_status(res) return res.json()
[docs] @retry(exceptions=IonQRetriableError, max_delay=60, backoff=2, jitter=1) def get_calibration_data( self, backend_name: str, limit: int | None = None ) -> Characterization | list[Characterization]: """Retrieve calibration data for a specified backend. Args: backend_name (str): The IonQ backend to fetch data for. limit (int, optional): Limit the number of results returned. Raises: IonQAPIError: When the API returns a non-200 status code. IonQRetriableError: When a retriable error occurs during the request. Returns: Characterization: An instance of Characterization containing the calibration data or a list of Characterization instances if multiple results are returned. """ params = {"limit": limit} if limit else None url = self.make_path("backends", backend_name, "characterizations") res = self.get_with_retry(url, headers=self.api_headers, params=params) exceptions.IonQAPIError.raise_for_status(res) chars = res.json().get("characterizations", []) return ( Characterization(chars[0]) if limit == 1 else [Characterization(item) for item in chars] )
[docs] @retry(exceptions=IonQRetriableError, max_delay=60, backoff=2, jitter=1) def get_results( self, results_url: str, sharpen: Optional[bool] = None, extra_query_params: Optional[dict] = None, ) -> dict: """Retrieve job results from the IonQ API. The returned JSON dict will only have data if job has completed. Args: results_url (str): The URL of the job results to retrieve. sharpen (bool): Supported if the job is debiased, allows you to filter out physical qubit bias from the results. extra_query_params (dict): Specify any parameters to include in the request Raises: IonQAPIError: When the API returns a non-200 status code. IonQRetriableError: When a retriable error occurs during the request. Returns: dict: A :mod:`requests <requests>` response :meth:`json <requests.Response.json>` dict. """ params = {} if sharpen is not None: params["sharpen"] = sharpen if extra_query_params is not None: warn( ( f"The parameter(s): {extra_query_params} is not checked by default " "but will be submitted in the request." ) ) params.update(extra_query_params) # Strip second API version (/v0.4/) req_path = re.sub(r"/v\d+\.\d+/", "", self.make_path(results_url), count=1) res = self.get_with_retry(req_path, headers=self.api_headers, params=params) exceptions.IonQAPIError.raise_for_status(res) # Use json.loads with object_pairs_hook to maintain order of JSON keys return json.loads(res.text, object_pairs_hook=OrderedDict)
[docs] def estimate_job( self, *, backend: str, oneq_gates: int, twoq_gates: int, qubits: int, shots: int, error_mitigation: bool = False, session: bool = False, job_type: str = "ionq.circuit.v1", ) -> JobEstimate: """Call GET /jobs/estimate … returns a cost/time prediction.""" params = { "type": job_type, "backend": backend.replace("ionq_qpu", "qpu"), "1q_gates": oneq_gates, "2q_gates": twoq_gates, "qubits": qubits, "shots": shots, "error_mitigation": str(error_mitigation).lower(), "session": str(session).lower(), } url = self.make_path("jobs", "estimate") res = self.get_with_retry(url, headers=self.api_headers, params=params) exceptions.IonQAPIError.raise_for_status(res) return JobEstimate(res.json())
[docs] @retry(exceptions=IonQRetriableError, tries=5) def post(self, *path_parts: str, json_body: dict | None = None) -> dict: """POST helper with IonQ headers + retry. Args: *path_parts (str): Path parts to append to the base URL. json_body (dict, optional): JSON body to send in the POST request. Raises: IonQAPIError: When the API returns a non-200 status code. IonQRetriableError: When a retriable error occurs during the request. Returns: dict: A :mod:`requests <requests>` response :meth:`json <requests.Response.json>` dict. """ url = self.make_path(*path_parts) res = requests.post(url, json=json_body, headers=self.api_headers, timeout=30) exceptions.IonQAPIError.raise_for_status(res) return res.json()
[docs] @retry(exceptions=IonQRetriableError, tries=3) def put(self, *path_parts: str, json_body: dict | None = None) -> dict: """PUT helper with IonQ headers + retry. Args: *path_parts (str): Path parts to append to the base URL. json_body (dict, optional): JSON body to send in the PUT request. Raises: IonQAPIError: When the API returns a non-200 status code. IonQRetriableError: When a retriable error occurs during the request. Returns: dict: A :mod:`requests <requests>` response :meth:`json <requests.Response.json>` dict. """ url = self.make_path(*path_parts) res = requests.put(url, json=json_body, headers=self.api_headers, timeout=30) exceptions.IonQAPIError.raise_for_status(res) return res.json()
[docs] class Characterization: """ Simple wrapper around the `/backends/<backend>/characterizations/<uuid>` payload. """ def __init__(self, data: dict) -> None: self._data = data # metadata @property def id(self) -> str: # pylint: disable=invalid-name """UUID of this characterization.""" return self._data["id"] @property def backend(self) -> str: """Backend name, e.g. `"qpu.aria-1"`.""" return self._data["backend"] @property def status(self) -> str: """Status of the characterization, e.g. `"available"`.""" return self._data["status"] @property def date(self) -> datetime: """Timestamp of the measurement (UTC).""" return datetime.fromisoformat(self._data["date"].replace("Z", "+00:00")) # qubit info @property def qubits(self) -> int: """Number of qubits available.""" return int(self._data["qubits"]) @property def connectivity(self) -> list[tuple[int, int]]: """Valid two-qubit gate pairs as a list of tuples.""" return [tuple(pair) for pair in self._data.get("connectivity", [])] # fidelity block @property def fidelity(self) -> dict: """Full fidelity dictionary (spam, 1q, 2q, ...).""" return self._data.get("fidelity", {}) @property def median_spam_fidelity(self) -> float | None: # convenience accessor """Median state-prep-and-measurement fidelity, if present.""" return self.fidelity.get("spam", {}).get("median") # timing block @property def timing(self) -> dict: """Dictionary of timing parameters (readout, reset, 1q, 2q, t1, t2).""" return self._data.get("timing", {}) def __repr__(self) -> str: parts = [ f"backend={self.backend}", f"id={self.id}", f"date={self.date.isoformat()}", f"qubits={self.qubits}", f"connectivity={self.connectivity}", f"fidelity={self.fidelity}", f"timing={self.timing}", ] return f"Characterization({', '.join(parts)})" def __eq__(self, other: object) -> bool: return isinstance(other, Characterization) and other._data == self._data
[docs] class JobEstimate: """ Wrapper for the payload returned by GET /jobs/estimate. """ def __init__(self, data: dict): # we keep the original dict just in case (for .to_dict()) self._raw = data # Flatten the interesting bits so they are attributes inputs = data.get("input_values", {}) self.backend: str | None = inputs.get("backend") self.oneq_gates: int | None = inputs.get("1q_gates") self.twoq_gates: int | None = inputs.get("2q_gates") self.qubits: int | None = inputs.get("qubits") self.shots: int | None = inputs.get("shots") self.error_mitigation: bool = inputs.get("error_mitigation") self.session: bool = inputs.get("session") # Core numeric results self.cost: float | None = data.get("estimated_cost") self.cost_unit: str | None = data.get("cost_unit") self.exec_time: float | None = data.get("estimated_execution_time") # seconds self.queue_time: float | None = data.get("current_predicted_queue_time") # sec # When was this generated? ts = data.get("estimated_at") # pylint: disable=invalid-name self.estimated_at: datetime | None = ( datetime.fromisoformat(ts) if isinstance(ts, str) else None ) # Optional structured pricing breakdown self.rate_information: dict = data.get("rate_information", {}) # convenience helpers @property def total_runtime(self) -> float | None: """Predicted queue + execution time, in seconds (if both are present).""" if self.exec_time is None or self.queue_time is None: return None return self.exec_time + self.queue_time
[docs] def to_dict(self) -> dict: """Return a shallow copy of the original JSON dict.""" return dict(self._raw)
def __repr__(self) -> str: parts = [ f"backend={self.backend}", f"qubits={self.qubits}", f"shots={self.shots}", f"cost={self.cost} {self.cost_unit}", f"exec_time={self.exec_time}s", f"queue_time={self.queue_time}s", ] return f"JobEstimate({', '.join(parts)})" def __eq__(self, other: object) -> bool: return isinstance(other, JobEstimate) and other._raw == self._raw
__all__ = ["IonQClient", "Characterization", "JobEstimate"]