Source code for qrisp.interface.qunicorn.backend_server

* Copyright (c) 2023 the Qrisp authors
* This program and the accompanying materials are made available under the
* terms of the Eclipse Public License 2.0 which is available at
* This Source Code may also be made available under the following Secondary
* Licenses when the conditions for such availability set forth in the Eclipse
* Public License, v. 2.0 are satisfied: GNU General Public License, version 2
* with the GNU Classpath Exception which is
* available at
* SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0

from flask import Flask, request, jsonify, make_response
import yaml
from datetime import date, datetime
import threading


This file sets up a server adhering to the interface specified by the Qunicorn middleware
developed in the SeQuenC project:

If you are a backend provider, we don't recommend using this server for production,
since it only exposes the necessities to run quantum circuits and nothing else.
Instead check out the Qunicorn GitHub:
And it's documentation on how to create new Pilots:


# Returns the hosts ip
def get_ip():
    import socket

    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        # doesn't even have to be reachable
        s.connect(("", 1))
        IP = s.getsockname()[0]
    except Exception:
        IP = ""
    return IP

[docs]class BackendServer: """ This class allows convenient setup of a server respecting the `Qunicorn<>`_ interface. Parameters ---------- run_func : function A function that receives a QuantumCircuit, an integer specifying the amount of shots and a token in the form of a string. It returns the counts as a dictionary of bitstrings. ip_address : str, optional The IP address of where the listening socket should be opened. By default, the IP address of the hosting machine will be used. port : int, optional The port on which to listen for requests. By default, 9010 will be used. is_simulator : bool, optional A bool specifying whether the results returned by this server are due to a simulator. The default is True. Examples -------- We create a server listening on the localhost IP address using a run function which prints the token and queries the QASM-simulator. :: def run_func(qasm_str, shots): from qiskit import QuantumCircuit #Convert to qiskit qiskit_qc = QuantumCircuit.from_qasm_str(qasm_str) from qiskit import Aer qiskit_backend = Aer.get_backend('qasm_simulator') #Run Circuit on the Qiskit backend return, shots = shots).result().get_counts() from qrisp.interface import BackendServer example_server = BackendServer(run_func, ip_address = "", port = 8080) example_server.start() """ def __init__(self, run_func , ip_address = None, port = None): = Flask(__name__) self.deployments = [] if ip_address is None: ip_address = get_ip() if port is None: port = 9010 self.port = port self.ip_address = ip_address = [] self.run_func = run_func self.is_simulator = True # To set up the server we route the relevant functions for the flask app. #This function receives the deployment data and saves it in the #deployment attribute'/deployments', methods=['POST']) def deployments(): # Receive the quantum circuit data from the request data = request.get_json() #Append the deployment data to the library self.deployments.append(data) #For the response dictionary response = { "id": len(self.deployments)-1, "deployedBy": { "id": 0, "name": "" }, "deployedAt": str(, "name": "" } response["programs"] = data["programs"] #Create response object response_object = make_response(jsonify(response), 201) return response_object #This function receives the deployment id of a job and starts the execution'/jobs', methods=['POST']) def jobs_post(): # Receive the quantum circuit data from the request data = request.get_json() #Extract the deployment deployment_id = data["deploymentId"] deployment = self.deployments[deployment_id] #Create the wrapper for asynchronous execution of the job def job_wrapper(): for i in range(len(deployment["programs"])): counts_res = self.run_func(deployment["programs"][i]["quantumCircuit"], data["shots"], data["token"]) job_dic["counts"].append(counts_res) # try: # for i in range(len(deployment["programs"])): # counts_res = self.run_func(deployment["programs"][i]["quantumCircuit"], # data["shots"], # data["token"]) # job_dic["counts"].append(counts_res) # except Exception as e: # job_dic["exception"] = e job_dic["end_time"] = str( #Create the thread run_thread = threading.Thread(target = job_wrapper) #Create the dictionary to hold information about the job job_dic = {} job_dic["deploymentId"] = deployment_id job_dic["counts"] = [] job_dic["shots"] = data["shots"] job_dic["exception"] = None job_dic["quantumCircuits"] = deployment["programs"] job_dic["run_thread"] = run_thread job_dic["start_time"] = str( job_dic["end_time"] = "-" #Start the thread run_thread.start() #Append the job to jobs attribute #Form the response response = { "id": len(, "jobName": "", "jobState": "running" } response_object = make_response(jsonify(response), 201) return response_object #This function retrieves information about a certain job. It is used by #the client to inquire the results of the job'/jobs/<int:job_id>/', methods=['GET']) def get_job(job_id): # Search for the job with the specified job_id if not job_id < len( return jsonify({"error": "Job not found"}), 404 #Get job information job =[job_id] #Find out wether the job finished thread = job["run_thread"] if not thread.is_alive(): state = "finished" else: state = "running" #Find out wether the job suceeded if job["exception"] is not None: state = "failed" response = { "code": 0, "status": "failed", "message": str(job["exception"]), "errors": {} } response_object = make_response(jsonify(response)) return response_object #Form the response response = { "id": job_id, "executedBy": { "id": 0, "name": "" }, "executedOn": { "id": 0, "numQubits": -1, "isSimulator": self.is_simulator, "isLocal": True, "provider": { "id": 0, "withToken": True, "supportedLanguages": { "id": 0, "providerId": "string", "name": "" }, "name": "" } }, "progress": 0, "state": state, "type": "", "startedAt": job["start_time"], "finishedAt": job["end_time"], "data": "", "results": [{ "id": 0, "circuit": job["quantumCircuits"][i], "results": job["counts"][i], "resultType": "COUNTS", "metaData": {} } for i in range(len(job["counts"]))], "parameters": "" } response_object = make_response(jsonify(response), 201) return response_object
[docs] def start(self): from waitress import serve #Set up the thread to run the server def wrapper(): serve(, host=self.ip_address, port=self.port) #, port=self.port) thr = threading.Thread(target=wrapper) thr.setDaemon(True) # Start the thread thr.start() import requests # Hold programm until the server answers while True: try: response = requests.get("http://" + self.ip_address + ":" + str(self.port) + "/jobs/0/") except: continue break