qci_client package

Submodules

qci_client.base module

Base class for API clients.

class qci_client.base.BaseApi(_bearer_info: dict = <factory>, api_token: str | None = None, set_bearer_token_on_init: bool = True, url: str | None = None, authorize: str | None = None, _authorize: str = 'authorize', _download: str = 'contents', _add_headers: ~typing.Dict[str, str] | None = None, _user_not_authorized: str = 'user not authorized', timeout: float | None = 300.0, debug: bool = False)

Bases: object

Base class for API clients.

api_token: str | None = None

property auth_url: str

Return the URL used for authorization.

authorize: str | None = None

debug: bool = False

get_bearer_token() → Response

Request new bearer token. (Not cached here, see set_bearer_token.)

property headers

Headers with cached bearer token.

property headers_without_connection_close

Headers with cached bearer token, but without connection closing.

property headers_without_token

Headers without cached bearer token.

is_bearer_token_expired() → bool

Is current time > ‘expires’ time. TODO: expires_in should be ‘expires_at’

static refresh_token(func) → Callable

Return a wrapper function that can check an auth token.

set_bearer_token() → None

Set bearer token from request.

set_bearer_token_on_init: bool = True

timeout: float | None = 300.0

url: str | None = None

qci_client.data_converter module

Functions for data conversion.

qci_client.data_converter.build_graph_data_from_get_file(file_parts: Generator, file_type: str) → dict

Return graph data built from a given graph file generator and type.

Args:

file_parts: File-part generator file_type: Type of the file

Returns:

Dictionary of graph data

qci_client.data_converter.compute_results_step_len(data: ndarray | list) → int

Compute the step length for “chunking” the providd data.

Args:

data: An numpy array or list of data

Returns:

The step length for “chunking” the data

qci_client.data_converter.data_to_json(data: ndarray | spmatrix | Graph | list, file_type: str, file_name: str | None = None, debug: bool = False) → dict

Converts data input into JSON string that can be passed to Qatalyst REST API :param data: object to be converted to JSON

• currently constraint_penalties require one value for each constraint

Parameters:

• file_type – one of [“graph”, “qubo”, “objective”, “constraints”, “constraint_penalties”, “rhs”, “hamiltonian”]

• file_name – Optional user specified name for file to be uploaded

• debug – Optional, if set to True, enables debug output (default = False for no debug output)

Returns:

string (JSON format)

Note:

could add support for matrices stored as lists

qci_client.data_converter.get_size(obj, seen=None) → int

Recursively finds size of objects

Parameters:

• obj – data object to recursively compute size of

• seen – takes a set and is used in the recursive step only to record whether an object has been counted yet.

Return int:

qci_client.data_converter.load_json_file(file_name: str) → dict

Load a utf-8-encoded json file into a dictionary.

Parameters:

file_name – name of the JSON file to load

Return dict:

loaded json file

qci_client.data_converter.multipart_file(data: dict, compress: bool = False) → Generator

Break file-to-upload’s data dictionary into chunks, formatting correctly with each returned chunk.

Parameters:

data – dict, object from data_to_json() function

Returns:

generator of (chunk, part_num) tuples

qci_client.qci_client module

QciClient Utility class for user interactions with QCI API

class qci_client.qci_client.JobStatus

Bases: object

Allowed jobs statuses.

CANCELLED = 'CANCELLED'

COMPLETED = 'COMPLETED'

ERROR = 'ERROR'

QUEUED = 'QUEUED'

RUNNING = 'RUNNING'

SUBMITTED = 'SUBMITTED'

class qci_client.qci_client.QciClient(_bearer_info: dict = <factory>, api_token: str | None = None, set_bearer_token_on_init: bool = True, url: str | None = None, authorize: str | None = None, _authorize: str = 'authorize', _download: str = 'contents', _add_headers: ~typing.Dict[str, str] | None = None, _user_not_authorized: str = 'user not authorized', timeout: float | None = 300.0, debug: bool = False, max_workers: int = 8, files: str = 'files', jobs: str = 'jobs', _supported_job_types: ~typing.List[str] | None = None)

Bases: BaseApi

Provides requests for QCIs public API as well as utility functions for creating requests and processing entire jobs.

Parameters:

• max_workers – int, number of threads for concurrent file download calls

• files – url path fragment to specify files API endpoint

• jobs – url path fragment to specify jobs API endpoint

• _supported_job_types – list of job_types accepted by jobs endpoint

build_job_body(job_type: str, qubo_file_id: str | None = None, graph_file_id: str | None = None, hamiltonian_file_id: str | None = None, objective_file_id: str | None = None, constraints_file_id: str | None = None, rhs_file_id: str | None = None, job_params: dict | None = None, job_name: str = 'job_0', job_tags: list | None = None) → dict

Constructs body for job submission requests :param job_type: one of _supported_job_types :param qubo_file_id: file id from files API for uploaded qubo :param graph_file_id: file id from files API for uploaded graph :param hamiltonian_file_id: file id from files API for uploaded hamiltonian :param objective_file_id: file id from files API for uploaded objective :param constraints_file_id: file id from files API for uploaded constraints :param rhs_file_id: file id from files API for uploaded rhs :param job_params: dict of additional params to be passed to job submission in “params” key :param job_name: user specified name for job submission :param job_tags: user specified labels for classifying and filtering user jobs after submission :note: Need to add validation for job parameters

delete_file(file_id: str) → dict

Parameters:

file_id – str - file_id of file to be deleted

Returns:

dict containing information about file deleted (or error)

files: str = 'files'

property files_url

Get files URL.

get_file(file_id: str, metadata: dict = None) → Generator

Return file contents as a generator. Each entry will be a dictionary of the form

{ ‘file_name’: ‘some_name’, ‘file_type’: ‘qubo’, ‘num_variables’: ‘n’, ‘part_num’: part number, ‘data’: [{‘i’: i, ‘j’: j, ‘val: val}, …, {…}] }

Parameters:

• file_id – str, a file id

• metadata – dict, metadata obtained from ‘get_metadata’

Returns:

generator

get_file_contents_url(file_id: str, part_num: int) → str

Get file contents URL with file ID and file part number.

get_file_id_url(file_id: str) → str

Get file URL with file ID.

get_file_metadata(file_id: str) → dict

Get file info. Use to determine file type and dimensions for matrix construction.

Args:

file_id: str, from {‘file_id’: file_id} returned by a GET call

Returns:

Dictionary of metadata

get_file_whole(file_id: str) → dict

Return file content with complete data field. Extends get_file and combines the iterator portion into one data list.

{ ‘file_name’: ‘some_name’, ‘file_type’: ‘qubo’, ‘num_variables’: ‘n’, ‘data’: [{‘i’: i, ‘j’: j, ‘val: val}, …, {…}] }

Parameters:

file_id – str, a file id

Returns:

dict, file content with metadata

get_job_id_url(job_id: str) → str

Get job URL with job ID.

get_job_metrics_provider_url(job_id: str) → str

Getjob metrics for provider using job ID.

get_job_metrics_url(job_id: str) → str

Get job metrics using job ID.

get_job_response(job_id: str, job_type: str) → dict

Get a response for a job by id and type, which may/may not be finished.

Parameters:

• job_id – ID of job

• job_type – type of job, one of []

Return dict:

loaded json file

get_job_status(job_id: str) → dict

Get the status of a job by its ID.

Args:

job_id: ID of job

Returns:

Response from GET call to API

get_job_statuses_url(job_id: str) → str

Get job status using job ID.

get_job_type_from_job_id(job_id: str) → str

Get job type from job ID.

Args:

job_id: ID of the job

Returns:

Type of the job

get_job_type_job_id_url(job_type: str, job_id: str) → str

Get job URL with job type and job ID.

get_job_type_url(job_type: str) → str

Get job URL with job type.

jobs: str = 'jobs'

property jobs_url

Get jobs URL.

list_files(username: str | None = None) → dict

Parameters:

username – Optional str - username (to search for files owned by the named user) mostly useful when run by users with administrator privileges (such as QCI users) who can see all files. When called by an administrator, the username parameter is used to restrict the list files returned to be only the files owned by the user specified in the username parameter. When run by non-privileged users, this parameter is truly optional because non-privileged users will only ever see lists of files that they created.

Returns:

dict containing list of files

max_workers: int = 8

merge_two_dicts(x, y)

Given two dictionaries, merge them into a new dict as a shallow copy.

print_job_log(message: str) → None

Formats a messages for updating user with a time stamp appended :param message: a string to be passed in print statement

process_job(job_type: str, job_body: dict, wait: bool = True) → dict

Parameters:

• job_type – the type of job being processed must be one of _supported_job_types

• job_body – formatted json dict for body of job submission request

• wait – bool indicating whether or not user wants to wait for job to complete

Returns:

if wait is True, then dict with job_info response and results file

(results is None if ERROR or CANCELLED)

if wait is False, then response dict from submitted job, which includes job

ID for subsequent retrieval

Note:

what else do we want to return with the results? response_id, obviously job_id

submit_job(json_dict: dict, job_type: str) → dict

Submit a job via a request to QCI public API.

Args:

json_dict: formatted json body that includes all parameters for the job job_type: one of the _supported_job_types

Returns:

Response from POST call to API

upload_file(data: dict | ndarray | Graph | spmatrix | list, file_type: str | None = None, file_name: str | None = None, compress: bool | None = False) → dict

Uploads either a formatted dict that can be uploaded directly or if supplied with a np.ndarray, nx.Graph, sp.spmatrix attempts to create formatted json. Must include file_type paramter must be provided if using any format other than dict for your upload.

Parameters:

• data – either a formatted dict which will compose upload body or a python object which will be used to create the data field in the formatted request body.

• file_type – str used in tandem with non-dict objects to create the request body. Optional if uploading a formatted dict which will compose the request body. Must be one of [“qubo”, “hamiltonian”, “graph”, “objective”, “constraints”, “rhs”].

• file_name – Optional str provided by user for identification of the file.

Returns:

dict with a single entry {“file_id”: “some_file_id”}.

Any open requests should be terminated promptly upon exit, say, due to an exception.

validate_job_type(job_type: str) → None

Checks if a provided job type is a supported job type.

Args:

job_type: a job type to validate

Returns:

None

Raises AssertionError if job_type is not one of the _supported_job_types.

zip_payload(payload: str) → bytes

Parameters:

payload – str - json contents of file to be zipped

“return: zipped request_body

qci_client.utils module

Utilities for qci-client.

Module contents

Automatic imports for qci-client.