# 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"""
import json
from collections import OrderedDict
from typing import Optional
from warnings import warn
import requests
from retry import retry
from . import exceptions
from .helpers import qiskit_to_ionq, get_user_agent
from .exceptions import IonQRetriableError
[docs]
class IonQClient:
"""IonQ API Client
Attributes:
_url(str): A URL base to use for API calls, e.g. ``"https://api.ionq.co/v0.3"``
_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=None, url=None, custom_headers=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):
"""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):
"""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)}"
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) -> 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):
"""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):
"""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):
"""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):
"""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) -> dict:
"""Retrieve calibration data for a specified backend.
Args:
backend_name (str): The IonQ backend to fetch data for.
Raises:
IonQAPIError: When the API returns a non-200 status code.
IonQRetriableError: When a retriable error occurs during the request.
Returns:
dict: A dictionary of an IonQ backend's calibration data.
"""
req_path = self.make_path(
"/".join(["characterizations/backends", backend_name[5:], "current"])
)
res = self._get_with_retry(req_path, headers=self.api_headers)
exceptions.IonQAPIError.raise_for_status(res)
return res.json()
[docs]
@retry(exceptions=IonQRetriableError, max_delay=60, backoff=2, jitter=1)
def get_results(
self,
job_id: str,
sharpen: Optional[bool] = None,
extra_query_params: Optional[dict] = None,
):
"""Retrieve job results 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.
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)
req_path = self.make_path("jobs", job_id, "results")
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)
__all__ = ["IonQClient"]