Usage

Quick Start

 import fastqsim

client = fastqsim.init()   # reads FASTQUBIT_SESSION_TOKEN from environment

 qasm = """
 OPENQASM 3.0;
 include "stdgates.inc";
 qubit[2] q;
 bit[2]   c;

 h  q[0];
 cx q[0], q[1];
 c = measure q;
 """

 # Synchronous - blocks until the job finishes
 job = client.run(qasm, shots=2048, asynchronous=False)
 result = job.result()

 print(job.job_id, job.status.value)   # e.g. "3fa8... completed"
 print(result.counts)                  # {"00": 1024, "11": 1024}

What’s New

• Server-side long-polling for wait() and synchronous run() to reduce client polling round-trips.

• Parallelized run_batch() submissions using a thread pool with max_parallel control.

• Session ID caching per client instance to avoid repeated resolution calls.

• Reuse of latest fetched terminal job snapshot to reduce redundant API reads.

Usage Scenarios

Synchronous Run

Block until the job completes and inspect results inline. Best for interactive notebooks and short circuits.

What it does: submits one circuit and waits until it reaches a terminal state. Use when: you want a simple request/response style flow. Returns: a completed Job and a Result with measurement data.

job = client.run(
  qasm,
  backend="qiskit",
  shots=1024,
  asynchronous=False,   # default is False - blocks here
)

result = job.result()
print(result.counts)      # {"00": 512, "11": 512}
print(job.execution_time_seconds)  # e.g. 0.34

Output shape: result.counts is a bitstring-to-count mapping. Common failure + fix: validation errors usually mean malformed QASM or invalid arguments; verify circuit syntax and shot count.

Async / Fire-and-Forget

Submit a job and do other work while it runs. Retrieve results later with wait() or get().

What it does: queues work without blocking the caller thread immediately. Use when: your application has parallel work while simulations run. Returns: an initial Job (queued/running), then a completed Job after wait().

job = client.run(qasm, shots=4096, asynchronous=True)
print(f"Queued: {job.job_id}")

# ... do other work ...

# Block until terminal, returns updated Job
completed_job = client.wait(job.job_id, timeout=300)
result = completed_job.result()
print(result.counts)

Output shape: first response contains job metadata; final response includes terminal status and result payload. Common failure + fix: timeout can occur on long jobs; increase timeout or handle JobTimeoutError and continue polling.

Batch Execution

Submit multiple circuits in parallel. All jobs share the same backend, device, and shots. Use client.wait() to block on all of them.

What it does: submits multiple circuits concurrently via a client thread pool. Use when: you need throughput for independent circuits. Returns: a list of Job objects from submission and a list of completed jobs after waiting.

circuits = [qasm_bell, qasm_ghz, qasm_qft]

# Submitted concurrently (up to max_parallel threads)
jobs = client.run_batch(
  circuits,
  shots=1024,
  max_parallel=3,   # controls client-side thread concurrency
)

print([j.job_id for j in jobs])

# Wait for all to complete
completed = client.wait([j.job_id for j in jobs], timeout=600)

for job in completed:
  print(job.job_id, job.result().counts)

Output shape: list of jobs where each job.result().counts has per-circuit measurement totals. Common failure + fix: partial failures may appear in mixed job lists; inspect each job status and handle failed ones individually.

Cancel a Job

Cancellation is idempotent. Cancelling an already-terminal job is a no-op and returns the current state.

What it does: records cancellation intent for a queued/running job. Use when: a queued/running simulation is no longer needed. Returns: an updated Job reflecting cancelled or existing terminal state.

job = client.run(qasm, asynchronous=True)

# Request cancellation
updated_job = client.cancel(job.job_id)
print(updated_job.status.value)   # "cancelled" or already-terminal status

# Cancel via the Job object directly
job2 = client.run(qasm, asynchronous=True)
job2.cancel()
print(job2.status.value)

Output shape: returns a job object with latest status fields. Common failure + fix: cancellation is not a hard kill for in-flight worker execution; poll until terminal status is confirmed.

Warning

Cancellation signals the platform to stop scheduling this job but does not kill an already-running worker pod. If a worker is mid-execution it may finish, but any terminal state updates are ignored once cancelled.

Deadline-Safe Polling

Use wait() with a timeout and catch JobTimeoutError for robust production workflows.

What it does: enforces a client-side deadline while waiting for job completion. Use when: you need bounded latency in production control flows. Returns: a completed Job on success or a timeout exception.

from fastqsim import JobTimeoutError, JobFailedError

job = client.run(qasm, asynchronous=True)

try:
  completed = client.wait(job.job_id, timeout=120)
  result = completed.result()
  print(result.counts)
except JobTimeoutError:
  print(f"Job {job.job_id} did not finish within 120 s")
  # Optionally cancel it
  client.cancel(job.job_id)
except JobFailedError as exc:
  print(f"Job failed: {exc}")

Output shape: success path returns terminal job state with result access. Common failure + fix: repeated timeouts indicate long or congested workloads; retry with larger timeout or switch to async orchestration.

Search Job History

Query your job history with optional filters. Returns up to limit jobs ordered by most-recently-updated first.

What it does: fetches historical jobs filtered by status/backend/date. Use when: you need observability, debugging context, or reporting. Returns: a list of Job records with metadata and timing fields.

from datetime import datetime, timezone, timedelta

# All completed jobs in the last 24 hours
yesterday = datetime.now(timezone.utc) - timedelta(days=1)

jobs = client.search(
  run_status="completed",
  start_date=yesterday,
  limit=20,
)

for job in jobs:
  print(job.job_id, job.status.value, job.execution_time_seconds)

# Filter by backend
qiskit_jobs = client.search(backend="qiskit", limit=50)

# Failed jobs for investigation
failed = client.search(run_status="failed", limit=10)
for job in failed:
  print(job.job_id, job.error_message)

Output shape: list of jobs sorted by recent update time. Common failure + fix: empty result sets are usually filter mismatch; relax filters or widen date ranges.

Statevector Retrieval

Request a full statevector simulation. The statevector is returned as a list of complex values.

What it does: runs a statevector simulation and reads amplitudes from results. Use when: you need full quantum state inspection, not only counts. Returns: List[complex] amplitudes via job.get_statevector().

job = client.run(
  qasm,
  shots=0,               # No measurement shots needed
  simulation_type="statevector",
  asynchronous=False,
)

sv = job.get_statevector()
print(f"State dimension: {len(sv)}")   # 2^n_qubits
print(f"|00> amplitude: {sv[0]:.4f}")
print(f"|11> amplitude: {sv[3]:.4f}")

Output shape: vector length is 2^n for n qubits. Common failure + fix: very large states may be stored externally; rely on job.get_statevector() artifact fetch behavior and avoid excessive qubits.

Warning

For large qubit counts the statevector may exceed the inline-response limit. In that case it is stored as an external artifact and get_statevector() fetches it automatically.

External Artifact Download

When job.has_external_artifact is True, the full result JSON was uploaded externally. job.result() fetches and merges it automatically.

job = client.run(large_circuit_qasm, shots=8192, asynchronous=False)

if job.has_external_artifact:
  # result() transparently downloads the artifact
  result = job.result()
  print(result.counts)
  if result.statevector:
    print(f"Statevector length: {len(result.statevector)}")
else:
  # Results fit inline - no extra download needed
  print(job.result_counts)

Density Matrix and MPS Simulation

Density matrix is useful for open-system or noisy simulation and supports all gates.

job = client.run(
  qasm,
  simulation_type="density_matrix",
  shots=1024,
  asynchronous=False,
)
result = job.result()
print(result.counts)
# Density matrix may be available in result.density_matrix

MPS (Matrix Product State) is efficient for low-entanglement circuits with many qubits.

job = client.run(
  large_qasm,
  simulation_type="mps",
  shots=2048,
  backend="qiskit",
  asynchronous=False,
)
print(job.result().counts)

CHP (stabilizer/Clifford) can be exponentially faster for Clifford-only circuits.

job = client.run(
  clifford_qasm,
  simulation_type="chp",
  shots=4096,
  asynchronous=False,
)
print(job.result().counts)

Error Handling

All SDK exceptions inherit from FastQSimError.

from fastqsim import (
  FastQSimConnectionError,
  AuthenticationError,
  QuotaExceededError,
  ValidationError,
  JobFailedError,
  JobTimeoutError,
  JobNotFoundError,
  JobCouldNotCancelError,
)

try:
  client = fastqsim.init()
except FastQSimConnectionError as exc:
  print(f"Cannot reach API: {exc}")
  raise SystemExit(1)
except AuthenticationError as exc:
  print(f"Auth failed - check FASTQUBIT_SESSION_TOKEN (fqp_...): {exc}")
  raise SystemExit(1)

try:
  job = client.run(
    invalid_qasm,          # Bad QASM
    shots=-5,              # Invalid shots
    asynchronous=False,
  )
except ValidationError as exc:
  print(f"Bad request: {exc}")

try:
  job = client.run(good_qasm, shots=1024, asynchronous=False)
  result = job.result()
except QuotaExceededError:
  print("Monthly quota exhausted - upgrade your plan")
except JobFailedError as exc:
  print(f"Simulation failed: {exc.reason}")
except JobTimeoutError as exc:
  print(f"Job {exc.job_id} timed out after {exc.timeout}s")
except JobNotFoundError as exc:
  print(f"Job not found: {exc}")

Job Metadata and Timing

Every Job object carries lifecycle timestamps and resource-usage metrics.

job = client.run(
  qasm,
  shots=1024,
  asynchronous=False,
  job_name="bell-state-test",
  tags={"experiment": "v3", "team": "qcomp"},
  metadata={"circuit_version": "2025-03-31"},
)

# Lifecycle timestamps (all datetime, UTC)
print(job.queue_enqueued_at)
print(job.execution_started_at)
print(job.execution_finished_at)
print(job.completed_at)

# Performance metrics
print(f"Wall time:   {job.execution_time_seconds:.3f}s")
print(f"CPU time:    {job.cpu_seconds_total:.3f}s")
print(f"Peak RAM:    {job.peak_memory_mb:.1f} MB")

# Billing telemetry
print(f"CPU millicore-s: {job.billing_cpu_millicore_seconds:.2f}")
print(f"Memory GB-s:     {job.billing_memory_gb_seconds:.4f}")

# Tags and metadata round-trip
print(job.tags)       # {"experiment": "v3", "team": "qcomp"}
print(job.metadata)   # {"circuit_version": "2025-03-31"}

# Refresh from server (for async jobs)
job.refresh()
print(job.status.value)