# Copyright 2025 qBraid
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Module defining QbraidJob class
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from qbraid_core.services.runtime import (
QuantumRuntimeClient,
QuantumRuntimeServiceRequestError,
)
from qbraid_core.services.runtime.schemas.result import BatchResult as CoreBatchResult
from qbraid_core.services.runtime.schemas.result import Result as CoreResult
from qbraid._logging import logger
from qbraid.runtime.enums import JobStatus
from qbraid.runtime.exceptions import JobStateError, QbraidRuntimeError
from qbraid.runtime.job import QuantumJob
from qbraid.runtime.result import BatchResult, Result, ResultDataType
from qbraid.runtime.result_data import ResultData
if TYPE_CHECKING:
import qbraid_core.services.runtime
from qbraid_core.services.runtime.schemas.job import Program
import qbraid.runtime
[docs]
class QbraidJob(QuantumJob):
"""Class representing a qBraid job."""
[docs]
def __init__(
self,
job_id: str,
device: qbraid.runtime.QbraidDevice | None = None,
client: qbraid_core.services.runtime.QuantumRuntimeClient | None = None,
**kwargs,
):
super().__init__(job_id, device, **kwargs)
self._device = device
self._client = client
self._compiled_program: Program | list[Program] | None = None
@property
def client(self) -> qbraid_core.services.runtime.QuantumRuntimeClient:
"""
Lazily initializes and returns the client object associated with the job.
If the job has an associated device with a client, that client is used.
Otherwise, a new instance of QuantumClient is created and used.
Returns:
QuantumClient: The client object associated with the job.
"""
if self._client is None:
self._client = self._device.client if self._device else QuantumRuntimeClient()
return self._client
def queue_position(self) -> int | None:
"""Return the position of the job in the queue."""
return self.metadata()["queuePosition"]
def status(self) -> JobStatus:
"""Return the status of the job / task , among the values of ``JobStatus``."""
terminal_states = JobStatus.terminal_states()
if self._cache_metadata.get("status") not in terminal_states:
job_model = self.client.get_job(self.id)
status = job_model.status
job_data = job_model.model_dump(exclude={"statusMsg"})
if job_model.statusMsg is not None:
status.set_status_message(job_model.statusMsg)
self._cache_metadata.update({**job_data, "status": status})
return self._cache_metadata["status"]
def program(self) -> Program | list[Program]:
"""Return the program submitted with the job.
For a batch job, returns one ``Program`` per circuit, positionally aligned
with the circuits. Single-circuit jobs return a bare ``Program``.
This is the same object ``metadata()["program"]`` holds, exposed with the
schema type ``qbraid_core`` already returns rather than only through the
untyped metadata dict.
"""
if self._cache_metadata.get("program") is None:
self._cache_metadata["program"] = self.client.get_job_program(self.id)
return self._cache_metadata["program"]
def metadata(self) -> dict[str, Any]:
"""Return the metadata regarding the job."""
self._cache_metadata.pop("job_id", None)
self.program()
return super().metadata()
def compiled_program(self) -> Program | list[Program] | None:
"""Return the vendor-compiled program that executed on the QPU.
Reveals the physical qubits selected and the logical-to-physical mapping
chosen by the vendor's compiler — information not recoverable from
measurement counts. ``format`` is ``"qasm3"`` for IQM and ``"quil"`` for
Rigetti.
A batch job compiles one program per circuit, so this returns a list
positionally aligned with the circuits, matching :meth:`program`.
Returns ``None`` when the job has no compiled program: the device does
not report one (simulators), the job has not completed, or it predates
compiled-program capture. Other request failures propagate.
The result is cached once found, since a compiled program never changes.
Absence is not cached — a job polled before it completed will pick one up
on a later call.
"""
if self._compiled_program is None:
try:
self._compiled_program = self.client.get_job_compiled_program(self.id)
except QuantumRuntimeServiceRequestError as err:
if err.status_code == 404:
return None
raise
return self._compiled_program
def cancel(self) -> None:
"""Attempt to cancel the job."""
if self.is_terminal_state():
raise JobStateError("Cannot cancel job in a terminal state.")
self.client.cancel_job(self.id)
logger.info("Cancel job request validated.")
try:
logger.info("Waiting for job to cancel...")
self.wait_for_final_state(timeout=3, poll_interval=1)
except TimeoutError:
pass
status = self.status()
if status not in {JobStatus.CANCELLED, JobStatus.CANCELLING}:
raise QbraidRuntimeError(f"Failed to cancel job. Current status: {status.name}")
logger.info("Success. Current status: %s", status.name)
def result(
self, timeout: int | None = None
) -> Result[ResultDataType] | BatchResult[ResultDataType]:
"""Return the results of the job.
For single-circuit jobs, returns a single :class:`Result`.
For batch jobs (``numCircuits > 1``), returns a :class:`BatchResult`.
"""
self.wait_for_final_state(timeout=timeout)
job_data = self.client.get_job(self.id)
success = job_data.status == JobStatus.COMPLETED
num_circuits = job_data.numCircuits or 1
if success:
raw_result = self.client.get_job_result(self.id)
else:
empty = CoreResult(
status=job_data.status,
cost=job_data.cost,
timeStamps=job_data.timeStamps,
resultData={},
)
if num_circuits > 1:
raw_result = CoreBatchResult(
status=job_data.status,
cost=job_data.cost,
timeStamps=job_data.timeStamps,
results=[empty] * num_circuits,
)
else:
raw_result = empty
def _build_result(core_result: CoreResult) -> Result[ResultDataType]:
data = ResultData.from_object(core_result, job_data.experimentType)
return Result[ResultDataType](
device_id=job_data.deviceQrn,
job_id=job_data.jobQrn,
success=core_result.status == JobStatus.COMPLETED,
data=data,
time_stamps=core_result.timeStamps,
cost=core_result.cost,
status=core_result.status,
**data.extra,
)
if isinstance(raw_result, CoreBatchResult):
per_circuit = [_build_result(r) for r in raw_result.results]
return BatchResult[ResultDataType](
device_id=job_data.deviceQrn,
job_id=job_data.jobQrn,
success=success,
results=per_circuit,
time_stamps=raw_result.timeStamps,
cost=raw_result.cost,
status=raw_result.status,
)
return _build_result(raw_result)