# Copyright (c) 2021, Qu & Co
# All rights reserved.
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright notice, this
# list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
"""
This module implements a class for interacting with the QUBEC REST API using the
requests module. Some of the requests are wrapped using the backoff decorator to deal
with potential API failures or disconnections
"""
# built-in
import json
from dataclasses import asdict, dataclass
from datetime import datetime, timedelta
from functools import wraps
from json.decoder import JSONDecodeError
from pprint import pformat
from typing import Union
# external
import backoff
import requests
# local
from .parameters import (
QUBEC_SDK_ENDPOINT,
QUBEC_SDK_PORT,
QUBEC_SDK_PROTOCOL,
SEPARATOR,
QubecSdkError,
)
from .tools import get_logger
logger = get_logger(__name__)
[docs]@dataclass
class QubecTokenData:
"""
JWT token information
"""
username: Union[str, None] = None
password: Union[str, None] = None
api_token: Union[str, None] = None
last_update: Union[datetime, None] = None
[docs]class QubecClient:
[docs] def __init__(self,
http_protocol: str,
endpoint: str,
port: int,
verbose: bool = True) -> None:
"""
Initialize a new instance of a QUBEC REST API client
Args:
http_protocol (str): the HTTP protocol to use for interacting with QUBEC, can
be either 'http' or 'https' endpoint (str): the QUBEC remote endpoint to query
port (int): The port to use when building the request
verbose (bool): Whether to print debugging information or not
"""
self.http_protocol = http_protocol
self.endpoint = endpoint
self.port = port
self.url = f"{http_protocol}://{endpoint}:{port}/qubec/v1"
self.token = QubecTokenData()
self.token_expiration = timedelta(minutes=15)
self.verbose = verbose
[docs] @backoff.on_exception(
backoff.expo,
(requests.exceptions.RequestException, JSONDecodeError),
max_tries=5,
jitter=backoff.full_jitter,
base=5,
)
def login(self, username: str, password: str):
"""
Login to a QUBEC session using valid user credentials
Args:
username (str): the username as registered on the Qu&Co website
password (str): the password chosen by the user at registration time
"""
logger.debug(SEPARATOR)
logger.debug(f"POST request to the endpoint: {self.url}/auth/login")
logger.debug("Request body")
form = {"username": username, "password": "*" * len(password)}
logger.debug(pformat(form, indent=4))
pwd_form = {"username": username, "password": password}
response = requests.post(f"{self.url}/auth/login", data=pwd_form)
content = response.json()
if response.status_code == 200 and content["status"] == "success":
self.token = QubecTokenData(
username=username,
password=password,
api_token=content["data"]["access_token"],
last_update=datetime.utcnow(),
)
else:
raise QubecSdkError("The credentials provided are not valid")
[docs] def logout(self):
"""
Reset all token information thus initiating a new, non authenticated
QUBEC session
"""
self.token = QubecTokenData()
[docs] @backoff.on_exception(
backoff.expo,
(requests.exceptions.RequestException, JSONDecodeError),
max_tries=5,
jitter=backoff.full_jitter,
base=5,
)
def current_user(self, token: str = None) -> dict:
"""
Get current user information from a valid QUBEC token either
stored or given as input
Args:
token (str): A valid, i.e. not expired, JWT token of the QUBEC platform
Returns:
A dictionary with the current logged in user
"""
token_bkp = asdict(self.token)
if token is not None:
self.token.api_token = token
self.token.last_update = datetime.utcnow()
response = requests.get(f"{self.url}/users/current",
headers=self._auth_header())
content = response.json()
if response.status_code == 200 and content["status"] == "success":
username = content["data"]["username"]
if username != self.token.username:
self.token = QubecTokenData(**token_bkp)
raise QubecSdkError(
"You are using a token associated to a different user. Logout and begin a new session."
)
return content["data"]
else:
self.token = QubecTokenData(**token_bkp)
raise QubecSdkError("The given token is not valid")
[docs] def refresh_session(self) -> None:
"""
Regenerate token information in the case that it has expired
"""
if self.token.api_token is None:
raise QubecSdkError("Login before executing any QUBEC submission")
if datetime.utcnow() - self.token.last_update >= self.token_expiration:
self.login(self.token.username, self.token.password)
[docs] @backoff.on_exception(
backoff.expo,
(requests.exceptions.RequestException, JSONDecodeError),
max_tries=5,
jitter=backoff.full_jitter,
base=5,
)
def submit_job(self, data: dict) -> dict:
"""
Submit a new QUBEC job. If an error occurs
in performing the requests due to a temporary disconnection, the request will
be retried up to 3 times
Sample output:
```
{
"status": "success",
"application": "qubec",
"version": "1.0",
"data" {
"job_id": "14eefd92-5aaf-43e6-9e0f-1f4784021155"
},
"message": "Job successfully submitted"
}
```
Args:
data (dict): Dictionary with all job specification parameters
Returns:
A dictionary with API response JSON
"""
logger.debug(SEPARATOR)
logger.debug(f"POST request to the endpoint: {self.url}/jobs/submit")
logger.debug("Request Body")
logger.debug(pformat(data, indent=4))
# schedule the job to QUBEC
response = requests.post(f"{self.url}/jobs/submit",
data=json.dumps(data),
headers=self._auth_header())
content = response.json()
if response.status_code != 200:
raise QubecSdkError(
f"Failed to schedule the job. Error message: {content['message']}"
)
else:
return content
[docs] @backoff.on_exception(backoff.expo,
Exception,
max_tries=5,
jitter=backoff.full_jitter,
base=5)
def progress_job(self, job_id: str) -> dict:
"""
Query a given job ID for progresses. If an error occurs
in performing the requests due to a temporary disconnection, the request will
be retried up to 3 times
Sample output:
```
{
"status": "success",
"application": "qubec",
"version": "1.0",
"data": {
"job_id": ,
"algorithm_type": "vqa",
"status": "completed",
"result": {
"optimal_value": 1.2345
"optimal_theta": 1.2345
"history_values": [1.2345, 1.2345],
"history_values_std": [1.2345, 1.2345],
"history_theta": [ [1.2345, 1.2345], [1.2345, 1.2345] ],
}
"error_msg": null,
"details": {
"start_date": 2020-10-10T00:00:00,
"wall_clock": 12345
}
}
}
```
Args:
job_id (str): the job identifier
Returns:
A dictionary with API response JSON
"""
response = requests.get(f"{self.url}/jobs/progress?job_id={job_id}",
headers=self._auth_header())
content = response.json()
if response.status_code != 200:
raise QubecSdkError(content["message"])
else:
return content
[docs] @backoff.on_exception(
backoff.expo,
(requests.exceptions.RequestException, JSONDecodeError),
max_tries=5,
jitter=backoff.full_jitter,
base=5,
)
def cancel_job(self, job_id: str) -> dict:
"""
Request a remote cancellation of a supposedly running job. If an error occurs
in performing the requests due to a temporary disconnection, the request will
be retried up to 3 times
Sample output:
```
{
"status": "success",
"application": "qubec",
"version": "1.0",
"data" {
"job_id": "14eefd92-5aaf-43e6-9e0f-1f4784021155"
},
"message": "Job successfully cancelled"
}
```
Args:
job_id (str): A string containing the unique job identifier
Returns:
A dictionary with API response JSON
Raises:
QubecSdk error if the response did not succeeded
"""
response = requests.get(f"{self.url}/jobs/cancel?job_id={job_id}",
headers=self._auth_header())
content = response.json()
if response.status_code != 200:
raise QubecSdkError(content["message"])
else:
return content
def _auth_header(self) -> str:
"""
Build the authentication header for authenticated requests
"""
return {"Authorization": f"Bearer {self.token.api_token}"}
[docs]def memoize(fn):
memoized_client: QubecClient = None
@wraps(fn)
def wrapper(*args, use_cached: bool = True):
if use_cached:
nonlocal memoized_client
if memoized_client is None:
memoized_client = fn(*args)
else:
memoized_client = fn(*args)
return memoized_client
return wrapper
[docs]@memoize
def get_client(
http_protocol: str = QUBEC_SDK_PROTOCOL,
endpoint: str = QUBEC_SDK_ENDPOINT,
port: int = QUBEC_SDK_PORT,
):
"""
Get a client for performing QUBEC REST API requests
Args:
http_protocol (str): the HTTP protocol to use for interacting with QUBEC, can be either 'http' or 'https'
endpoint (str): the QUBEC remote endpoint to query
port (int): The port to use when building the request
use_cached (bool): Whether to use or not a memoized instance of the QUBEC client previously initialized
Returns:
An instance of a QUBEC client either new and not authenticated or previously initialized
and potentially already authenticated
"""
client = QubecClient(http_protocol, endpoint, port)
return client