API Reference

prefect_qiskit.runtime

This module provides an abstraction layer for managing the execution of Qiskit Primitives. Execution is managed by a Prefect task behind the scenes, enhancing system error resilience.

The runtime supports both synchronous and asynchronous execution of primitives. Asynchronous execution is particularly efficient for I/O-bound tasks, typical in client-server remote computing scenarios.

Users must supply Prefect credential blocks for the specific vendor to run on the target hardware.

QuantumRuntime

Bases: Block

Prefect Block used to execute Qiskit Primitives on quantum computing resources. Quantum Runtime is a vendor agnostic implementation of primitive execution.

Warning

Cache doesn't work as expected until Qiskit 2.1 release. See qiskit/#12963.

Attributes:

Name	Type	Description
resource_name	str	Name of a quantum computing resource available with your credentials. Input value is validated and an error is raised when the resource name is not found.
credentials	QuantumCredentialsT	Credentials to access the quantum computing resource from a target vendor.
enable_job_analytics	bool	Enable quantum job analytics. When analytics is enabled, each primitive execution will create a table artifact 'job-metrics' that reports various execution metrics the vendor provides as a job metadata.
max_retry	int	Maximum number of primitive execution retry on failure. Primitive execution raises an error after all executions fail or the error is not retryable.
retry_delay	int	Standby time in seconds before creating new execution task on failure. This setting is applied only if the maximum retry is nonzero.
timeout	int | None	Primitive execution timeout in seconds. Execution task will raise TaskRunTimeoutError after timeout. If the maximum number of reruns has not been reached, the error is also suppressed and new execution task is created.
execution_cache	bool	Cache primitive execution result in a local file system. A cache key is computed from the input Primitive Unified Blocs, options, and resource name. Overhead of the primitive operands evaluation is incurred.

Example

Load stored Quantum Runtime and run Qiskit sampler primitive.

from prefect_qiskit.runtime import QuantumRuntime
from qiskit.circuit import QuantumCircuit
from qiskit.transpiler.preset_passmanagers import generate_preset_pass_manager

runtime = QuantumRuntime.load("BLOCK_NAME")

vqc = QuantumCircuit(2)
vqc.h(0)
vqc.cx(0, 1)
vqc.measure_all()

pm = generate_preset_pass_manager(optimization_level=2, target=runtime.get_target())

result = runtime.sampler(pm.run([vqc]))

async_estimator async

async_estimator(estimator_pubs, options=None, tags=None)

Asynchronously run estimator primitive.

Parameters:

Name	Type	Description	Default
estimator_pubs	list[SamplerPubLike]	Operands of estimator primitive.	required
options	dict | None	Vendor-specific primitive options.	None
tags	list[str] | None	Arbitrary labels to add to the Primitive task tags of execution.	None

Returns:

Type	Description
PrimitiveResult	Qiskit PrimitiveResult object.

async_get_target async

async_get_target()

Asynchronously get Qiskit Target of specific backend.

Returns:

Type	Description
Target	Qiskit Target object.

async_sampler async

async_sampler(sampler_pubs, options=None, tags=None)

Asynchronously run sampler primitive.

Parameters:

Name	Type	Description	Default
sampler_pubs	list[SamplerPubLike]	Operands of sampler primitive.	required
options	dict | None	Vendor-specific primitive options.	None
tags	list[str] | None	Arbitrary labels to add to the Primitive task tags of execution.	None

Returns:

Type	Description
PrimitiveResult	Qiskit PrimitiveResult object.

estimator

estimator(estimator_pubs, options=None, tags=None)

Run estimator primitive.

Parameters:

Name	Type	Description	Default
estimator_pubs	list[EstimatorPubLike]	Operands of estimator primitive.	required
options	dict | None	Vendor-specific primitive options.	None
tags	list[str] | None	Arbitrary labels to add to the Primitive task tags of execution.	None

Returns:

Type	Description
PrimitiveResult	Qiskit PrimitiveResult object.

get_target

get_target()

Get Qiskit Target of specific backend.

Returns:

Type	Description
Target	Qiskit Target object.

sampler

sampler(sampler_pubs, options=None, tags=None)

Run sampler primitive.

Parameters:

Name	Type	Description	Default
sampler_pubs	list[SamplerPubLike]	Operands of sampler primitive.	required
options	dict | None	Vendor-specific primitive options.	None
tags	list[str] | None	Arbitrary labels to add to the Primitive task tags of execution.	None

Returns:

Type	Description
PrimitiveResult	Qiskit PrimitiveResult object.

prefect_qiskit.vendors

This module implements credentials for each quantum computing vendor, following the Prefect CredentialsBlock structure. Users can create and store these credentials on the Prefect server.

The block must implement the .get_client method, which returns a client instance for the vendor-specific API, adhering to the AsyncRuntimeClientInterface protocol. This client is used by the primitive runner task and is not exposed externally. Although users can access the client instance via this method, it should be considered a private class and may be subject to future API changes without deprecation.

IBMQuantumCredentials

Bases: CredentialsBlock

Block used to manage runtime service authentication with IBM Quantum.

Attributes:

Name	Type	Description
api_key	SecretStr	API key to access IBM Quantum Platform.
crn	str	Cloud resource name to identify the service to use.
auth_url	HttpUrl | None	A custom endpoint URL for IAM authentication.
runtime_url	HttpUrl | None	A custom endpoint URL for Qiskit Runtime API.

Example

Load stored IBM Quantum credentials:

from prefect_qiskit.vendors.ibm_quantum import IBMQuantumCredentials

quantum_credentials = IBMQuantumCredentials.load("BLOCK_NAME")

get_client

get_client()

Get IBM Runtime REST API client.

QiskitAerCredentials

Bases: CredentialsBlock

Block used to emulate a runtime with the Qiskit Aer simulator. This block provides a virtual system configuration to produce a circuit ISA.

Attributes:

Name	Type	Description
num_qubits	int	Number of qubits involved in the simulation. The host machine or computing node must have enough memory to represent qubit states, otherwise workflow execution may hang up.
basis_gates	list[str]	A list of quantum gate opcode to form a basis gate set. The 'delay', 'measure', and 'reset' instruction are included without being specified here.
noise	bool	Simulate noise during quantum gate operation. The depolarizing error is considered for gates, and the assignment error is considered for measurements. See the API doc for NoiseModel in Qiskit Aer for more details.
coupling_map_type	Literal['full', 'ring', 'linear', 'grid', 'hexagonal', 'heavy-hex', 'heavy_square']	Topology of coupling map. Actual connections are automatically generated from the selected topology.

Example

Load stored Qiskit Aer credentials:

from prefect_qiskit.vendors.qiskit_aer import QiskitAerCredentials

quantum_credentials = QiskitAerCredentials.load("BLOCK_NAME")

get_client

get_client()

Get a mock runtime client running the Aer simulator.

prefect_qiskit.primitives

This module manages the execution of Primitives using Prefect.

Vendor-specific job handling is encapsulated within Prefect tasks, following a common job handling pattern in Prefect.

PrimitiveJob

Bases: JobBlock

Prefect-style definition of a single primitive job.

Although the job handling logic is backend-agnostic, users can provide vendor-specific credentials and execution options to communicate with target hardware via the runtime client API.

Attributes:

Name	Type	Description
resource_name	str	A target hardware name available with the given credentials.
credentials	CredentialsBlock	A vendor specific credentials that provides a runtime client API.
primitive_blocs	list[SamplerPub] | list[EstimatorPub]	Canonical Qiskit Primitive Unified Blocs data.
options	dict | None	A vendor specific execution options for primitives.
program_type	Literal['estimator', 'sampler']	Reserved primitive name to run on runtime client API.
job_watch_poll_interval	float	The amount of time to wait between runtime client API calls during monitoring.

trigger async

trigger()

Triggers a job run in an external service and returns a PrimitiveJobRun object to track the execution of the run.

PrimitiveJobRun

Bases: BaseModel, JobRun

A handler of running Primitive Job.

Tip

You can directly retrieve the result of the job by using Job ID.

from prefect_qiskit.primitives import PrimitiveJobRun
from prefect_qiskit.vendors.ibm_quantum import IBMQuantumCredentials

job = PrimitiveJobRun(
    job_id="d0a61oqorbds73c39gt0",
    credentials=IBMQuantumCredentials.load("my-ibm-client"),
)
result = await job.fetch_result()

Attributes:

Name	Type	Description
job_id	str	The unique identifier of the running job.
job_watch_poll_interval	float	The amount of time to wait between Runtime API calls during monitoring.
credentials	CredentialsBlock	A vendor specific credentials that provides a runtime client API.

fetch_metrics async

fetch_metrics()

Fetch JobMetrics object from server.

fetch_result async

fetch_result()

Fetch Qiskit PrimitiveResult object from server.

wait_for_completion async

wait_for_completion()

Asynchronously wait for the job to complete.

run_primitive async

run_primitive(*, primitive_blocs, program_type, resource_name, credentials, enable_analytics=True, options=None)

This function implements a Prefect task to manage the execution of Qiskit Primitives on an abstract layer, providing built-in execution failure protection.

It accepts PUBs and options, submitting this data to quantum computers via the vendor's API. The response is formatted into a Qiskit PrimitiveResult instance.

If the job is successful and the enable_analytics flag is set, the task will analyze job metadata, create a job-metrics artifact, and save it on the Prefect server.

Parameters:

Name	Type	Description	Default
primitive_blocs	list[SamplerPub] | list[EstimatorPub]	Input sampler PUBs.	required
program_type	Literal['sampler', 'estimator']	Primitive ID.	required
resource_name	str	Quantum computing resource.	required
credentials	CredentialsBlock	Runtime API credentials.	required
enable_analytics	bool	Set True to generate table artifact of job metrics.	True
options	dict | None	Vendor-specific primitive options.	None

Returns:

Type	Description
PrimitiveResult	Qiskit PrimitiveResult object.

prefect_qiskit.models

This module abstracts the runtime client interface and common data structures.

The Quantum Runtime implementation for Prefect adheres to the Qiskit programming model for quantum primitives, which is vendor hardware agnostic.

Primitives utilize Primitive Unified Blocs (PUBs) along with an execution options dictionary. Execution options can specify a vendor-specific schema based on supported features, such as built-in error mitigations.

This module defines the protocol for API adapters that Prefect primitives can use. The protocol does not specify actual communication methods, allowing hardware vendors to implement any protocols, such as RESTful or gRPC, to interact with their hardware. All methods are asynchronous to leverage Prefect's native async support.

JOB_STATUS module-attribute

JOB_STATUS = Literal['QUEUED', 'RUNNING', 'COMPLETED', 'FAILED', 'CANCELLED']

Reserved status string of runtime job.

AsyncRuntimeClientInterface

Bases: Protocol

Interface abstraction of runtime client API version 1.

A hardware vendor must provide a credential Block along with its API client that implements all the interfaces defined with this protocol.

check_resource_available async

check_resource_available(resource_name)

Asynchronously check whether the target resource is currently available.

Parameters:

Name	Type	Description	Default
resource_name	str	Name of quantum computing resource.	required

Returns:

Type	Description
bool	True when resource is currently available.

get_job_metrics async

get_job_metrics(job_id)

Asynchronously get metrics of Job.

Parameters:

Name	Type	Description	Default
job_id	str	Unique identifier of the job.	required

Returns:

Type	Description
JobMetrics	Execution metrics of the job when available.

get_job_status async

get_job_status(job_id)

Asynchronously get status string of Job ID.

Note

This function raises error immediately when a job failure occurs. A client must inspect whether the job is retriable or not. In case the job is retriable, a downstream primitive runner will resubmit the same job data to the server.

Parameters:

Name	Type	Description	Default
job_id	str	Unique identifier of the job.	required

Raises:

Type	Description
RuntimeJobFailure	When job execution fails.

Returns:

Type	Description
JOB_STATUS	Job status literal.

get_primitive_result async

get_primitive_result(job_id)

Asynchronously get Qiskit PrimitiveResult object from complete Job ID.

Parameters:

Name	Type	Description	Default
job_id	str	Unique identifier of the job.	required

Returns:

Type	Description
PrimitiveResult	Qiskit PrimitiveResult object.

get_resources async

get_resources()

Asynchronously list available resource names for this client.

Returns:

Type	Description
list[str]	A list of resource names under this account.

get_target async

get_target(resource_name)

Asynchronously get Qiskit Target for a specific computing resource.

Parameters:

Name	Type	Description	Default
resource_name	str	Name of quantum computing resource.	required

Returns:

Type	Description
Target	Qiskit Target.

run_primitive async

run_primitive(program_id, inputs, resource_name, options)

Asynchronously send Primitive Blocs via Runtime API and return Job ID.

Parameters:

Name	Type	Description	Default
program_id	Literal['sampler', 'estimator']	Type of primitive to use.	required
inputs	list[SamplerPub] | list[EstimatorPub]	Qiskit Pub payloads that the primitive API will digest.	required
resource_name	str	Name of quantum computing resource.	required
options	dict[str, Any]	Execution options.	required

Returns:

Type	Description
str	Job ID.

JobMetrics

Bases: NamedTuple

A canonical job metrics template. Vendor can provide following data if available. Provided information is consumed by the primitive runner to create a job metrics artifact.

Attributes:

Name	Type	Description
qpu_usage	float | None	Duration that a quantum processor was actively running.
timestamp_created	datetime | None	Time when the user job was created in the queue.
timestamp_started	datetime | None	Time when the execution started on the control server.
timestamp_completed	datetime | None	Time when the QPU execution completed on the control server.

prefect_qiskit.exceptions

This module defines custom exceptions.

RuntimeJobFailure

Bases: Exception

A custom exception to enable fine-grained control of Prefect task retries.

All exceptions raised within the implementation of AsyncRuntimeClientInterface must be type cast into this exception. This exception includes a retry flag, allowing the Prefect primitive runner to determine whether to retry the task upon encountering an exception.

For instance, if a user creates invalid PUB data, the task execution is unlikely to succeed. In such cases, the error from PUB validation should be type cast into RuntimeJobFailure with retry=False.

__init__

__init__(reason, job_id=None, error_code=None, retry=True)

Define exception.

Parameters:

Name	Type	Description	Default
reason	str	A human readable explanation of the failure.	required
job_id	str | None	Failed Job ID.	None
error_code	str | None	Vendor-specific error code.	None
retry	bool	Either retry the primitive execution or not.	True