Exclusive

ExclusiveRemoteTask dataclass

ExclusiveRemoteTask(
    _task_ir: QuEraTaskSpecification | None,
    _metadata: Dict[str, ParamType],
    _parallel_decoder: ParallelDecoder | None,
    _http_handler: HTTPHandlerABC = HTTPHandler(),
    _task_id: str | None = None,
    _task_result_ir: QuEraTaskResults | None = None,
)

Bases: CustomRemoteTaskABC

pull

pull(poll_interval: float = 20)

Blocking pull to get the task result. poll_interval is the time interval to poll the task status. Please ensure that it is relatively large, otherwise the server could get overloaded with queries.

Source code in .venv/lib/python3.12/site-packages/bloqade/analog/task/exclusive.py

def pull(self, poll_interval: float = 20):
    """
    Blocking pull to get the task result.
    poll_interval is the time interval to poll the task status.
    Please ensure that it is relatively large, otherwise
    the server could get overloaded with queries.
    """

    while True:
        if self._task_result_ir.task_status is QuEraTaskStatusCode.Unsubmitted:
            raise ValueError("Task ID not found.")

        if self._task_result_ir.task_status in [
            QuEraTaskStatusCode.Completed,
            QuEraTaskStatusCode.Partial,
            QuEraTaskStatusCode.Failed,
            QuEraTaskStatusCode.Unaccepted,
            QuEraTaskStatusCode.Cancelled,
        ]:
            return self

        status = self.status()
        if status in [QuEraTaskStatusCode.Completed, QuEraTaskStatusCode.Partial]:
            self._task_result_ir = self._http_handler.fetch_results(self._task_id)
            return self

        time.sleep(poll_interval)

HTTPHandler

HTTPHandler(
    zapier_webhook_url: str = None,
    zapier_webhook_key: str = None,
    vercel_api_url: str = None,
)

Bases: HTTPHandlerABC

Source code in .venv/lib/python3.12/site-packages/bloqade/analog/task/exclusive.py

def __init__(
    self,
    zapier_webhook_url: str = None,
    zapier_webhook_key: str = None,
    vercel_api_url: str = None,
):
    self.zapier_webhook_url = zapier_webhook_url or os.environ["ZAPIER_WEBHOOK_URL"]
    self.zapier_webhook_key = zapier_webhook_key or os.environ["ZAPIER_WEBHOOK_KEY"]
    self.verrcel_api_url = vercel_api_url or os.environ["VERCEL_API_URL"]

fetch_results

fetch_results(task_id: str)

Fetch the task results from the AirTable.

Parameters:

Name	Type	Description	Default
task_id	str	The task id to be queried.	required

returns response: The response from the AirTable. used for error handling

Source code in .venv/lib/python3.12/site-packages/bloqade/analog/task/exclusive.py

def fetch_results(self, task_id: str):
    response = request(
        "GET",
        self.verrcel_api_url,
        params={
            "searchPattern": task_id,
            "magicToken": self.zapier_webhook_key,
            "useRegex": False,
        },
    )
    if response.status_code != 200:
        print(f"HTTP request failed with status code: {response.status_code}")
        print("HTTP responce: ", response.text)
        return None

    response_data = response.json()
    # Get "matched" from the response
    matches = response_data.get("matches", None)
    # The return is a list of dictionaries
    # Verify if the list contains only one element
    if matches is None:
        print("No task found with the given ID.")
        return None
    elif len(matches) > 1:
        print("Multiple tasks found with the given ID.")
        return None
    record = matches[0]
    if record.get("status") == "Completed":
        googledoc = record.get("resultsFileUrl")

        # convert the preview URL to download URL
        googledoc = convert_preview_to_download(googledoc)
        res = get(googledoc)
        res.raise_for_status()
        data = res.json()

        task_results = QuEraTaskResults(**data)
    return task_results

query_task_status

query_task_status(task_id: str)

Query the task status from the AirTable.

Parameters:

Name	Type	Description	Default
task_id	str	The task id to be queried.	required

returns response: The response from the AirTable. used for error handling

Source code in .venv/lib/python3.12/site-packages/bloqade/analog/task/exclusive.py

def query_task_status(self, task_id: str):
    response = request(
        "GET",
        self.verrcel_api_url,
        params={
            "searchPattern": task_id,
            "magicToken": self.zapier_webhook_key,
            "useRegex": False,
        },
    )
    if response.status_code != 200:
        return "HTTP Request Failed."
    response_data = response.json()
    # Get "matched" from the response
    matches = response_data.get("matches", None)
    # The return is a list of dictionaries
    # Verify if the list contains only one element
    if matches is None:
        print("No task found with the given ID.")
        return "Task searching Failed"
    elif len(matches) > 1:
        print("Multiple tasks found with the given ID.")
        return "Task searching Failed"

    record = matches[0]

    # Extract the status from the first dictionary
    status = record.get("status")

    if status == "Failed validation":
        googledoc = record.get("resultsFileUrl")

        # convert the preview URL to download URL
        googledoc = convert_preview_to_download(googledoc)
        res = get(googledoc)
        res.raise_for_status()
        data = res.json()
        # get the "statusCode" and "message" from the data and print them out.
        status_code = data.get("statusCode", "NA")
        message = data.get("message", "NA")
        print(
            f"Task validation failed with status code: {status_code}, message: {message}"
        )

    return status

submit_task_via_zapier

submit_task_via_zapier(
    task_ir: QuEraTaskSpecification,
    task_id: str,
    task_note: str,
)

Submit a task and add task_id to the task fields for querying later.

Parameters:

Name	Type	Description	Default
task_ir	QuEraTaskSpecification	The task to be submitted.	required
task_id	str	The task id to be added to the task fields.	required

returns response: The response from the Zapier webhook. used for error handling

Source code in .venv/lib/python3.12/site-packages/bloqade/analog/task/exclusive.py

def submit_task_via_zapier(
    self, task_ir: QuEraTaskSpecification, task_id: str, task_note: str
):
    # implement http request logic to submit task via Zapier
    request_options = dict(params={"key": self.zapier_webhook_key, "note": task_id})

    # for metadata, task_ir in self._compile_single(shots, use_experimental, args):
    json_request_body = task_ir.json(exclude_none=True, exclude_unset=True)

    request_options.update(data=json_request_body)
    response = request("POST", self.zapier_webhook_url, **request_options)

    if response.status_code == 200:
        response_data = response.json()
        submit_status = response_data.get("status", None)
        return submit_status
    else:
        print(f"HTTP request failed with status code: {response.status_code}")
        print("HTTP responce: ", response.text)
        return "HTTP Request Failed"

HTTPHandlerABC

fetch_results abstractmethod

fetch_results(task_id: str)

Fetch the task results from the AirTable.

Parameters:

Name	Type	Description	Default
task_id	str	The task id to be queried.	required

returns response: The response from the AirTable. used for error handling

Source code in .venv/lib/python3.12/site-packages/bloqade/analog/task/exclusive.py

@abc.abstractmethod
def fetch_results(task_id: str):
    """Fetch the task results from the AirTable.

    args:
        task_id: The task id to be queried.

    returns
        response: The response from the AirTable. used for error handling

    """

    ...

query_task_status abstractmethod

query_task_status(task_id: str)

Query the task status from the AirTable.

Parameters:

Name	Type	Description	Default
task_id	str	The task id to be queried.	required

returns response: The response from the AirTable. used for error handling

Source code in .venv/lib/python3.12/site-packages/bloqade/analog/task/exclusive.py

@abc.abstractmethod
def query_task_status(task_id: str):
    """Query the task status from the AirTable.

    args:
        task_id: The task id to be queried.

    returns
        response: The response from the AirTable. used for error handling

    """
    ...

submit_task_via_zapier abstractmethod

submit_task_via_zapier(
    task_ir: QuEraTaskSpecification, task_id: str
)

Submit a task and add task_id to the task fields for querying later.

Parameters:

Name	Type	Description	Default
task_ir	QuEraTaskSpecification	The task to be submitted.	required
task_id	str	The task id to be added to the task fields.	required

returns response: The response from the Zapier webhook. used for error handling

Source code in .venv/lib/python3.12/site-packages/bloqade/analog/task/exclusive.py

@abc.abstractmethod
def submit_task_via_zapier(task_ir: QuEraTaskSpecification, task_id: str):
    """Submit a task and add task_id to the task fields for querying later.

    args:
        task_ir: The task to be submitted.
        task_id: The task id to be added to the task fields.

    returns
        response: The response from the Zapier webhook. used for error handling

    """
    ...