# This code is part of Qiskit.
#
# (C) Copyright Alpine Quantum Technologies GmbH 2023
#
# This code is licensed under the Apache License, Version 2.0. You may
# obtain a copy of this license in the LICENSE.txt file in the root directory
# of this source tree or at [http://www.apache.org/licenses/LICENSE-2.0](http://www.apache.org/licenses/LICENSE-2.0).
#
# Any modifications or derivative works of this code must retain this
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.
from typing import Final
from uuid import UUID
import aqt_connector
import httpx
from aqt_connector import ArnicaApp
from aqt_connector.exceptions import (
InvalidJobIDError,
JobNotFoundError,
NotAuthenticatedError,
RequestError,
UnknownServerError,
)
from aqt_connector.models.arnica.jobs import JobStatus as AQTJobStatus
from aqt_connector.models.arnica.response_bodies.jobs import JobState
from qiskit.providers import JobV1
from qiskit.providers.jobstatus import JobStatus as QiskitJobStatus
from qiskit.result import Result
from qiskit_aqt_provider._cloud.job_metadata import CloudJobMetadata
from qiskit_aqt_provider._transformers import partial_qiskit_result_dict
from qiskit_aqt_provider.exceptions import (
AQTApiError,
AQTCredentialsError,
AQTJobFailedError,
AQTJobInvalidStateError,
AQTRequestError,
AQTValueError,
)
from qiskit_aqt_provider.persistence import JobSnapshot, JobStore, persist_job
[docs]
class CloudJob(JobV1):
"""A job representing the execution of one or more circuits on an AQT cloud resource."""
STATUS_MAPPING: Final = {
AQTJobStatus.CANCELLED: QiskitJobStatus.CANCELLED,
AQTJobStatus.ERROR: QiskitJobStatus.ERROR,
AQTJobStatus.FINISHED: QiskitJobStatus.DONE,
AQTJobStatus.ONGOING: QiskitJobStatus.RUNNING,
AQTJobStatus.QUEUED: QiskitJobStatus.QUEUED,
}
def __init__(self, arnica: ArnicaApp, api_client: httpx.Client, properties: CloudJobMetadata) -> None:
"""Initializes a cloud job with the given properties, HTTP client, and Arnica app instance."""
self._arnica = arnica
self._api_client = api_client
self._properties = properties
self._latest_state: JobState = properties.initial_state
super().__init__(None, str(properties.job_id))
[docs]
def submit(self) -> None:
"""Do not call — submission is handled by the backend.
This job object represents an execution that has already been submitted via backend.run(). Calling submit() is
invalid and always raises a RuntimeError.
Raises:
RuntimeError: Job submission is performed by backend.run().
"""
raise RuntimeError("Job is already submitted via backend.run()")
[docs]
def persist(self, *, store: JobStore | None = None) -> None:
"""Persist this submitted job for restoration in a later process."""
persist_job(
JobSnapshot(
job_id=UUID(self.job_id()),
backend_kind="cloud",
backend_name=self._properties.backend_name,
shots=self._properties.shots,
memory=self._properties.memory,
circuits=self._properties.circuits,
),
store,
)
[docs]
def result(
self,
*,
timeout: float | None = None,
wait: float = 5,
) -> Result:
"""Blocks until the job finishes processing then returns the result.
If an error occurs, the remaining circuits are not executed and the whole job is marked as failed.
Raises:
APIError: the operation failed on the target resource.
AQTJobInvalidStateError: if the job was cancelled.
AQTJobFailedError: if the job failed with an error.
qiskit.providers.exceptions.JobTimeoutError: If the job does not reach a final state before the specified
timeout.
Returns:
The combined result of all circuit evaluations.
"""
self.wait_for_final_state(timeout=timeout, wait=wait)
if self._latest_state.status == AQTJobStatus.ERROR:
error_message = self._latest_state.message or "Unknown error"
raise AQTJobFailedError(f"Job failed: {error_message}")
if self._latest_state.status != AQTJobStatus.FINISHED:
raise AQTJobInvalidStateError(
f"Unable to retrieve result for job {self.job_id()}. Job is {self._latest_state.status.lower()}."
)
result_dict = {
"backend_name": self._properties.backend_name,
"job_id": self.job_id(),
"success": True,
"results": [],
}
for circuit_index, circuit in enumerate(self._properties.circuits):
samples = self._latest_state.result[circuit_index]
result_dict["results"].append(
partial_qiskit_result_dict(
samples, circuit, shots=self._properties.shots, memory=self._properties.memory
)
)
return Result.from_dict(result_dict)
[docs]
def status(self) -> QiskitJobStatus:
"""Return the status of the job, among the values of :class:`~qiskit.providers.JobStatus`.
Raises:
AQTCredentialsError: if the user is not authenticated and no access token is available.
AQTCredentialsError: If the provided token is invalid or expired.
AQTRequestError: If there is a network-related error during the request.
AQTValueError: If the job with the specified ID does not exist.
AQTValueError: If the provided job ID is not valid.
AQTApiError: If the Arnica API encounters an internal error.
AQTApiError: For any other unexpected errors.
Returns:
qiskit.providers.JobStatus: The current status of the job.
"""
if self._latest_state.status in (AQTJobStatus.FINISHED, AQTJobStatus.ERROR, AQTJobStatus.CANCELLED):
return self.STATUS_MAPPING[self._latest_state.status]
try:
self._latest_state = aqt_connector.fetch_job_state(self._arnica, UUID(self.job_id()))
except NotAuthenticatedError as e:
raise AQTCredentialsError(str(e)) from e
except RequestError as e:
raise AQTRequestError(str(e)) from e
except (UnknownServerError, RuntimeError) as e:
raise AQTApiError(str(e)) from e
except (InvalidJobIDError, JobNotFoundError) as e:
raise AQTValueError(str(e)) from e
return self.STATUS_MAPPING.get(self._latest_state.status, QiskitJobStatus.QUEUED)