Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Unitary-Hack][Sub-quera] Doc string for submission/quera.py #980

Closed
99 changes: 99 additions & 0 deletions src/bloqade/submission/quera.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,6 @@
"""Module to host QueraBackend class that hosts different functionalities
representing the backend of the QuEra system(s)."""

from pydantic.v1 import PrivateAttr
from bloqade.submission.base import SubmissionBackend, ValidationError
from bloqade.submission.ir.task_specification import (
Expand All @@ -12,6 +15,34 @@


class QuEraBackend(SubmissionBackend):
"""Class to represent QuEra backend.

Attributes:
api_hostname (str): API of the host.
qpu_id (str): QPU id.
api_stage (str): API version. Defaults to "v0"
virtual_queue (Optional[str]) = Virtual queue in QuEra backend.
Defaults to `None`.
proxy (Optional[str]): proxy for Quera backend. Defaults to `None`.
# Sigv4Request arguments
region (Optional[str]): Sigv4Request argument for region of QuEra backend.
Defaults to `None`.
access_key (Optional[str]): Sigv4Request argument representing the access
key required to access QuEra backend. Defaults to `None`.
secret_key (Optional[str]): Sigv4Request argument representing the secret
key required to access QuEra backend. Defaults to `None`.
session_token (Optional[str]): Sigv4Request argument representing the session
token of the QuEra backend. Defaults to `None`.
session_expires (Optional[int]) = Sigv4Request argument representing the
expiration timestamp for the session on QuEra backend. Defaults to `None`.
role_arn (Optional[str]) = Role of the user accessing QuEra backend.
Defaults to `None`.
role_session_name (Optional[str]) = Role session name for QuEra backend.
Defaults to `None`.
profile (Optional[str]) = User profile for QuEra backend.
Defaults to `None`.
"""

api_hostname: str
qpu_id: str
api_stage: str = "v0"
Expand Down Expand Up @@ -42,27 +73,85 @@ def queue_api(self):
return self._queue_api

def get_capabilities(self, use_experimental: bool = False) -> QuEraCapabilities:
"""Get the capabilities of the QuEra backend.

Args:
use_experimental (bool): Whether to use experimental capabilities of
the QuEra system. Defaults to `False`.

Returns:
capabilities (QuEraCapabilities): capabilities of the selected
QuEra backend.
"""
try:
return QuEraCapabilities(**self.queue_api.get_capabilities())
except BaseException:
return super().get_capabilities(use_experimental)

def submit_task(self, task_ir: QuEraTaskSpecification) -> str:
"""Submit a task to the QuEra backend.

Args:
task_ir (QuEraTaskSpecification): task IR to be
executed on the QuEra backend.

Returns:
task_id (str): Task id as a result of executing
IR on the QuEra backend.
"""
return self.queue_api.submit_task(
task_ir.json(by_alias=True, exclude_none=True, exclude_unset=True)
)

def task_results(self, task_id: str) -> QuEraTaskResults:
"""Get the status of a task previously submitted to the QuEra backend
by using the task id.

Args:
task_id (str): task id after executing program on the QuEra backend.

Returns:
task_result (QuEraTaskResults):
Final result of a task previously submitted by using the task id.

Note:
This is a blocking call, meaning the function will not
return till the task has stopped.
"""
return QuEraTaskResults(**self.queue_api.poll_task_results(task_id))

def cancel_task(self, task_id: str):
"""Cancels a task previously submitted to the QuEra backend.

Args:
task_id (str): task id after executing program on the QuEra backend.
"""
self.queue_api.cancel_task_in_queue(task_id)

def task_status(self, task_id: str) -> QuEraTaskStatusCode:
"""Get the status of a task previously submitted to the QuEra backend
by using the task id.

Args:
task_id (str): task id after executing program on the QuEra backend.

Returns:
task_status (QuEraTaskStatusCode):
status of a task previously by using the task id.
"""
return_body = self.queue_api.get_task_status_in_queue(task_id)
return QuEraTaskStatusCode(return_body)

def validate_task(self, task_ir: QuEraTaskSpecification):
"""Validates the task submitted to the QuEra backend.

Args:
task_ir (QuEraTaskSpecification): task IR to be
executed on the QuEra backend.

Raises:
ValidationError: For tasks that fail validation.
"""
try:
self.queue_api.validate_task(
task_ir.json(by_alias=True, exclude_none=True, exclude_unset=True)
Expand All @@ -73,6 +162,16 @@ def validate_task(self, task_ir: QuEraTaskSpecification):
def update_credential(
self, access_key: str = None, secret_key: str = None, session_token: str = None
):
"""Update the credentials

Args:
access_key (Optional[str]): Sigv4Request argument representing the access
key required to access QuEra backend. Defaults to `None`
secret_key (Optional[str]): Sigv4Request argument representing the secret
key required to access QuEra backend. Defaults to `None`
session_token (Optional[str]): Sigv4Request argument representing the
session token of the QuEra backend. Defaults to `None`
"""
if secret_key is not None:
self.secret_key = secret_key
if access_key is not None:
Expand Down
Loading