Backends

The divi.backends module provides interfaces for running quantum circuits on different backends, from local simulators to cloud-based quantum hardware.

Backend Architecture

All backends implement the CircuitRunner interface, providing a consistent API for circuit execution. This abstraction allows switching between different execution environments without changing quantum program code.

class CircuitRunner(shots)[source]

Bases: ABC

A generic interface for anything that can “run” quantum circuits.

__init__(shots)[source]

property shots

Get the number of measurement shots for circuit execution.

Returns:: Number of shots configured for this runner.
Return type:: int

abstract property supports_expval: bool: Whether the backend supports expectation value measurements.

abstract property is_async: bool

Whether the backend executes circuits asynchronously.

Returns:

True if the backend returns a job ID and requires polling: for results (e.g., QoroService). False if the backend returns results immediately (e.g., ParallelSimulator).

Return type:

bool

abstract submit_circuits(circuits, **kwargs)[source]

Submit quantum circuits for execution.

This abstract method must be implemented by subclasses to define how circuits are executed on their respective backends (simulator, hardware, etc.).

Parameters:

circuits (dict[str, str]) – Dictionary mapping circuit labels to their OpenQASM string representations.
**kwargs – Additional backend-specific parameters for circuit execution.

Returns:

For synchronous backends, contains results directly.: For asynchronous backends, contains a job_id that can be used to fetch results later.

Return type:

ExecutionResult

Execution Results

All CircuitRunner.submit_circuits() methods return an ExecutionResult object, which provides a unified interface for handling both synchronous and asynchronous backend responses.

class ExecutionResult(results=None, job_id=None)[source]

Bases: object

Result container for circuit execution.

This class provides a unified return type for all CircuitRunner.submit_circuits() methods. For synchronous backends, it contains the results directly. For asynchronous backends, it contains the job_id that can be used to fetch results later.

The class is frozen (immutable) to ensure data integrity. Use the with_results() method to create a new instance with results populated from an async ExecutionResult.

results

For sync backends or after fetching: List of result dictionaries, each containing ‘label’ and ‘results’ keys. Format: [{“label”: str, “results”: dict}, …]

Type:: list[dict] | None

job_id

For async backends: Job identifier that can be used to poll and retrieve results from the backend.

Type:: str | None

Examples

>>> # Synchronous backend
>>> result = ExecutionResult(results=[{"label": "circuit_0", "results": {"00": 100}}])
>>> result.is_async()
False

>>> # Asynchronous backend
>>> result = ExecutionResult(job_id="job-12345")
>>> result.is_async()
True
>>> # After fetching results
>>> result = backend.get_job_results(result)
>>> result.results is not None
True

results: list[dict] | None = None

str, “results”: dict}, …]

Type:: Results for synchronous backends
Type:: [{“label”

job_id: str | None = None: Job identifier for asynchronous backends.

__init__(results=None, job_id=None)

is_async()[source]

Check if this result represents an async job.

Returns:

True if job_id is not None and results are None (async backend),: False otherwise (sync backend or results already fetched).

Return type:

bool

with_results(results)[source]

Create a new ExecutionResult with results populated.

This method creates a new instance with results set, effectively converting an async ExecutionResult to a completed one.

Parameters:

results (list[dict]) – The job results to populate.

Returns:

A new ExecutionResult instance with results populated: and job_id preserved.

Return type:

ExecutionResult

Core Backend Classes

class ParallelSimulator(n_processes=None, shots=5000, simulation_seed=None, qiskit_backend=None, noise_model=None, _deterministic_execution=False)[source]

Bases: CircuitRunner

A parallel wrapper around Qiskit’s AerSimulator using Qiskit’s built-in parallelism.

Parameters:

n_processes (int | None, optional) – Number of parallel processes to use for transpilation and simulation. If None, defaults to half the available CPU cores (min 2, max 8). Controls both transpilation parallelism and execution parallelism. The execution parallelism mode (circuit or shot) is automatically selected based on workload characteristics.
shots (int, optional) – Number of shots to perform. Defaults to 5000.
simulation_seed (int, optional) – Seed for the random number generator to ensure reproducibility. Defaults to None.
qiskit_backend (Backend | Literal["auto"] | None, optional) – A Qiskit backend to initiate the simulator from.
passed (If "auto" is)
circuit. (the best-fit most recent fake backend will be chosen for the given)
None (Defaults to)
simulation. (resulting in noiseless)
noise_model (NoiseModel, optional) – Qiskit noise model to use in simulation. Defaults to None.

__init__(n_processes=None, shots=5000, simulation_seed=None, qiskit_backend=None, noise_model=None, _deterministic_execution=False)[source]

A parallel wrapper around Qiskit’s AerSimulator using Qiskit’s built-in parallelism.

Parameters:

n_processes (int | None, optional) – Number of parallel processes to use for transpilation and simulation. If None, defaults to half the available CPU cores (min 2, max 8). Controls both transpilation parallelism and execution parallelism. The execution parallelism mode (circuit or shot) is automatically selected based on workload characteristics.
shots (int, optional) – Number of shots to perform. Defaults to 5000.
simulation_seed (int, optional) – Seed for the random number generator to ensure reproducibility. Defaults to None.
qiskit_backend (Backend | Literal["auto"] | None, optional) – A Qiskit backend to initiate the simulator from.
passed (If "auto" is)
circuit. (the best-fit most recent fake backend will be chosen for the given)
None (Defaults to)
simulation. (resulting in noiseless)
noise_model (NoiseModel, optional) – Qiskit noise model to use in simulation. Defaults to None.

set_seed(seed)[source]

Set the random seed for circuit simulation.

Parameters:: seed (int) – Seed value for the random number generator used in simulation.

property n_processes: int

Get the current number of parallel processes.

Returns:: Number of parallel processes configured.
Return type:: int

property supports_expval: bool: Whether the backend supports expectation value measurements.

property is_async: bool: Whether the backend executes circuits asynchronously.

submit_circuits(circuits)[source]

Submit multiple circuits for parallel simulation using Qiskit’s built-in parallelism.

Uses Qiskit’s native batch transpilation and execution, which handles parallelism internally.

Parameters:

circuits (dict[str, str]) – Dictionary mapping circuit labels to OpenQASM string representations.

Returns:

Contains results directly (synchronous execution).: Results are in the format: [{“label”: str, “results”: dict}, …]

Return type:

ExecutionResult

static estimate_run_time_single_circuit(circuit, qiskit_backend, **transpilation_kwargs)[source]

Estimate the execution time of a quantum circuit on a given backend, accounting for parallel gate execution.

Parameters:

circuit (str) – The quantum circuit to estimate execution time for as a QASM string.
qiskit_backend (Union[Backend, Literal['auto']]) – A Qiskit backend to use for gate time estimation.

Returns:

Estimated execution time in seconds.

Return type:

float

static estimate_run_time_batch(circuits=None, precomputed_durations=None, n_qpus=5, **transpilation_kwargs)[source]

Estimate the execution time of a quantum circuit on a given backend, accounting for parallel gate execution.

Parameters:

circuits (list[str]) – The quantum circuits to estimate execution time for, as QASM strings.
precomputed_durations (list[float]) – A list of precomputed durations to use.
n_qpus (int) – Number of QPU nodes in the pre-supposed cluster we are estimating runtime against.

Returns:

Estimated execution time in seconds.

Return type:

float

class QoroService(auth_token=None, config=None, polling_interval=3.0, max_retries=5000)[source]

Bases: CircuitRunner

A client for interacting with the Qoro Quantum Service API.

This class provides methods to submit circuits, check job status, and retrieve results from the Qoro platform.

Initializes the QoroService client.

Parameters:

auth_token (str | None, optional) – The authentication token for the Qoro API. If not provided, it will be read from the QORO_API_KEY in a .env file.
config (JobConfig | None, optional) – A JobConfig object containing default job settings. If not provided, a default configuration will be created.
polling_interval (float, optional) – The interval in seconds for polling job status. Defaults to 3.0.
max_retries (int, optional) – The maximum number of retries for polling. Defaults to 5000.

__init__(auth_token=None, config=None, polling_interval=3.0, max_retries=5000)[source]

Initializes the QoroService client.

Parameters:

auth_token (str | None, optional) – The authentication token for the Qoro API. If not provided, it will be read from the QORO_API_KEY in a .env file.
config (JobConfig | None, optional) – A JobConfig object containing default job settings. If not provided, a default configuration will be created.
polling_interval (float, optional) – The interval in seconds for polling job status. Defaults to 3.0.
max_retries (int, optional) – The maximum number of retries for polling. Defaults to 5000.

property supports_expval: bool: Whether the backend supports expectation value measurements.

property is_async: bool: Whether the backend executes circuits asynchronously.

test_connection()[source]

Test the connection to the Qoro API.

Sends a simple GET request to verify that the API is reachable and the authentication token is valid.

Returns:: The response from the API ping endpoint.
Return type:: requests.Response
Raises:: requests.exceptions.HTTPError – If the connection fails or authentication is invalid.

fetch_qpu_systems()[source]

Get the list of available QPU systems from the Qoro API.

Return type:: list[QPUSystem]
Returns:: List of QPUSystem objects.

submit_circuits(circuits, ham_ops=None, job_type=None, override_config=None)[source]

Submit quantum circuits to the Qoro API for execution.

This method first initializes a job and then sends the circuits in one or more chunks, associating them all with a single job ID.

Parameters:

circuits (dict[str, str]) – Dictionary mapping unique circuit IDs to QASM circuit strings.
ham_ops (str | None, optional) – String representing the Hamiltonian operators to measure, semicolon-separated. Each term is a combination of Pauli operators, e.g. “XYZ;XXZ;ZIZ”. If None, no Hamiltonian operators will be measured.
job_type (JobType | None, optional) – Type of job to execute (e.g., SIMULATE, EXECUTE, EXPECTATION, CIRCUIT_CUT). If not provided, the job type will be determined from the service configuration.
override_config (JobConfig | None, optional) – Configuration object to override the service’s default settings. If not provided, default values are used.

Raises:

ValueError – If more than one circuit is submitted for a CIRCUIT_CUT job, or if any circuit is not valid QASM.
requests.exceptions.HTTPError – If any API request fails.

Returns:

Contains job_id for asynchronous execution. Use the job_id: to poll for results using backend.poll_job_status() and get_job_results().

Return type:

ExecutionResult

delete_job(execution_result)[source]

Delete a job from the Qoro Database.

Parameters:: execution_result (ExecutionResult) – An ExecutionResult instance with a job_id to delete.
Returns:: The response from the API.
Return type:: requests.Response
Raises:: ValueError – If the ExecutionResult does not have a job_id.

cancel_job(execution_result)[source]

Cancel a job on the Qoro Service.

Parameters:

execution_result (ExecutionResult) – An ExecutionResult instance with a job_id to cancel.

Returns:

The response from the API. Use response.json() to get: the cancellation details (status, job_id, circuits_cancelled).

Return type:

requests.Response

Raises:

ValueError – If the ExecutionResult does not have a job_id.
requests.exceptions.HTTPError – If the cancellation fails (e.g., 403 Forbidden, or 409 Conflict if job is not in a cancellable state).

get_job_results(execution_result)[source]

Get the results of a job from the Qoro Database.

Parameters:

execution_result (ExecutionResult) – An ExecutionResult instance with a job_id to fetch results for.

Returns:

A new ExecutionResult instance with results populated.

Return type:

ExecutionResult

Raises:

ValueError – If the ExecutionResult does not have a job_id.
requests.exceptions.HTTPError – If the job results are not available (e.g., job is still running) or if the request fails.

poll_job_status(execution_result, loop_until_complete=False, on_complete=None, verbose=True, progress_callback=None)[source]

Get the status of a job and optionally execute a function on completion.

Parameters:

execution_result (ExecutionResult) – An ExecutionResult instance with a job_id to check.
loop_until_complete (bool) – If True, polls until the job is complete or failed.
on_complete (Callable, optional) – A function to call with the final response object when the job finishes.
verbose (bool, optional) – If True, prints polling status to the logger.
progress_callback (Callable, optional) – A function for updating progress bars. Takes (retry_count, status).

Returns:

The current job status.

Return type:

JobStatus

Raises:

ValueError – If the ExecutionResult does not have a job_id.

Job Management

class JobConfig(shots=None, qpu_system=None, use_circuit_packing=None, tag='default', force_sampling=False)[source]

Bases: object

Configuration for a Qoro Service job.

shots: int | None = None: Number of shots for the job.

qpu_system: QPUSystem | str | None = None: The QPU system to use, can be a string or a QPUSystem object.

use_circuit_packing: bool | None = None: Whether to use circuit packing optimization.

tag: str = 'default': Tag to associate with the job for identification.

force_sampling: bool = False: Whether to force sampling instead of expectation value measurements.

override(other)[source]

Creates a new config by overriding attributes with non-None values.

This method ensures immutability by always returning a new JobConfig object and leaving the original instance unmodified.

Parameters:: other (JobConfig) – Another JobConfig instance to take values from. Only non-None attributes from this instance will be used for the override.
Return type:: JobConfig
Returns:: A new JobConfig instance with the merged configurations.

__post_init__()[source]: Sanitizes and validates the configuration.

__init__(shots=None, qpu_system=None, use_circuit_packing=None, tag='default', force_sampling=False)

class JobStatus[source]

Status of a job on the Qoro Service.

PENDING = 'PENDING': Job is queued and waiting to be processed.

RUNNING = 'RUNNING': Job is currently being executed.

COMPLETED = 'COMPLETED': Job has finished successfully.

FAILED = 'FAILED': Job execution encountered an error.

CANCELLED = 'CANCELLED': Job was cancelled before completion.

class JobType[source]

Type of job to execute on the Qoro Service.

EXECUTE = 'EXECUTE': Execute circuits on real quantum hardware (sampling mode only).

SIMULATE = 'SIMULATE': Simulate circuits using cloud-based simulation services (sampling mode).

EXPECTATION = 'EXPECTATION': Compute expectation values for Hamiltonian operators (simulation only).

CIRCUIT_CUT = 'CIRCUIT_CUT': Automatically decompose large circuits that wouldn’t fit on a QPU.