qci_client

qci_client.qci_client

QciClient Utility class for user interactions with QCI API

class qci_client.qci_client.JobStatus[source]

Bases: object

Allowed jobs statuses.

class qci_client.qci_client.QciClient(url: str | None = None, api_token: str | None = None, access_tokens: str = 'auth/v1/access-tokens', set_bearer_token_on_init: bool = True, timeout: float | None = None, debug: bool = False, files: str = 'optimization/v1/files', jobs: str = 'optimization/v1/jobs', max_workers: int = 8, compress: bool = False)[source]

Bases: BaseApi

Provides requests for QCi’s public API for running optimization problems on Dirac devices, including file uploads/downloads and submitting/retrieving entire jobs.

Accepts same parameters as qci_client.base.BaseApi as well as:

Parameters:

files – url path fragment to specify files API endpoint
jobs – url path fragment to specify jobs API endpoint
max_workers – number of threads for concurrent file download calls
compress – compress file metadata and parts before uploading

Constructs body for job submission requests

Parameters:

job_type – one of _supported_job_types
job_params – dict of params to be passed to job submission in “params” key
qubo_file_id – file id from files API for uploaded qubo
graph_file_id – file id from files API for uploaded graph
hamiltonian_file_id – file id from files API for uploaded hamiltonian
objective_file_id – file id from files API for uploaded objective
constraints_file_id – file id from files API for uploaded constraints
polynomial_file_id – file id from files API for uploaded polynomial
job_name – user specified name for job submission
job_tags – user specified labels for classifying and filtering user jobs after submission

Returns:

None

delete_file(file_id: str) → dict[source]

Parameters:: file_id – str - file_id of file to be deleted
Returns:: dict containing information about file deleted (or error)

download_file(*, file_id: str) → dict[source]: Download file (metadata and then parts concurrently).

property files_url: Get files URL.

get_file_contents_url(file_id: str, part_num: int) → str[source]: Get file contents URL with file ID and file part number.

get_file_id_url(file_id: str) → str[source]: Get file URL with file ID.

get_job_allocations_url() → str[source]: Get job allocations

get_job_id_url(job_id: str) → str[source]: Get job URL with job ID.

get_job_response(job_id: str, job_type: str) → dict[source]

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[source]

Get the status of a job by its ID.

Parameters:: job_id – ID of job
Returns:: Response from GET call to API

get_job_status_url(job_id: str) → str[source]: Get job status using job ID.

get_job_type_from_job_id(job_id: str) → str[source]

Get job type from job ID.

Parameters:: job_id – ID of the job
Returns:: Type of the job

property headers_without_authorization: dict: HTTP headers without bearer token.

property jobs_url: Get jobs URL.

list_files(username: str | None = None) → dict[source]

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

print_job_log(message: str) → None[source]: 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[source]

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 ERRORED 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(job_body: dict, job_type: str) → dict[source]

Submit a job via a request to QCI public API.

Parameters:

job_body – 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(file: dict) → dict[source]: Upload file (metadata and then parts concurrently). Returns dict with file ID.

validate_job_type(job_type: str) → None[source]

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

Parameters:: 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[source]

Zip contents of json file

Parameters:: payload – str - json contents of file to be zipped
Returns:: zipped request_body

qci_client.utilities

Utilities for Python client for files REST API for optimization service.

qci_client.utilities.file_part_generator(*, file: dict, compress: bool) → Generator[source]

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

Parameters:

file – file to break up into file parts
compress – whether or not file parts are to be compressed

Returns:

generator of (part_body, part_number) tuples

qci_client.utilities.get_patch_request_body(*, file: dict) → PartPatchRequestBody[source]: Format part body.

qci_client.utilities.get_post_request_body(*, file: dict) → InputMetadataPostRequestBody | ResultsMetadataPostRequestBody[source]: Format metadata body.

qci_client.utilities.zip_payload(*, payload: PartPatchRequestBody) → bytes[source]

Parameters:: payload – str - json contents of file to be zipped
Returns:: zipped request_body

qci_client.data_converter

Functions for data conversion.

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

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

Parameters:: data – An numpy array or list of data
Returns:: The step length for “chunking” the data

qci_client.data_converter.data_to_json(file: dict, debug: bool = False) → dict[source]

Converts data in file input into JSON-serializable dictionary that can be passed to Qatalyst REST API

Parameters:

file – file dictionary whose data of type numpy.ndarray, scipy.sparse.spmatrix, or networkx.Graph is to be converted
debug – Optional, if set to True, enables debug output (default = False for no debug output)

Returns:

file dictionary with JSON-serializable data

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

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.base

Base class for API clients.

class qci_client.base.BaseApi(url: str | None = None, api_token: str | None = None, access_tokens: str = 'auth/v1/access-tokens', set_bearer_token_on_init: bool = True, timeout: float | None = None, debug: bool = False)[source]

Bases: object

Base class for clients to QCi APIs, especially the authorization layer.

Parameters:

url – url basepath to API endpoint, including scheme, if None, then falls back to QCI_API_URL environment variable
api_token – refresh token for authenticating to API, if None, then falls back to QCI_TOKEN environment variable
access_tokens – url path fragment to specify access-tokens API endpoint
set_bearer_token_on_init – flag to turn on/off access token retrieval on object initialization
timeout – number of seconds before timing out requests, None waits indefinitely
debug – flag to turn on/off debug prints

property auth_url: str: Return the URL used for authorization.

get_bearer_token() → Response[source]: 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[source]: Is current time > ‘expires’ time.

static refresh_token(func) → Callable[source]: Return a wrapper function that can check an auth token.

set_bearer_token() → None[source]: Set bearer token from request.