"""
FastQSim HTTP client.
Wraps all communication with the FastQubit REST API. Zero knowledge of
quantum backends, parsers, or circuit objects — accepts only OpenQASM 3.0
strings and returns Job / Result / Session objects.
SDK version 0.1.1 — pod-token-only authentication (fqp_).
"""
from __future__ import annotations
import base64
import json
import os
import time
import warnings
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from typing import Any, Dict, List, Optional, Union
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from .exceptions import (
AuthenticationError,
FastQSimConnectionError,
JobCouldNotCancelError,
JobDoesNotContainCountsError,
JobFailedError,
JobNotCompleteError,
JobNotFoundError,
JobStatevectorTooLargeError,
JobTimeoutError,
QuotaExceededError,
UnauthorizedAccessError,
ValidationError,
)
from .qtypes import Job, JobStatus, Result, Session
# SDK version — bump major on breaking auth model change.
SDK_VERSION = "0.1.1"
# Valid simulation types — mirrors SimulationType enum but kept as a set here
# so the client never needs to import types (avoids circular deps).
_VALID_SIMULATION_TYPES = {"statevector", "density_matrix", "mps", "chp"}
_DEFAULT_POLL_INTERVAL_SECONDS = 0.5
_POLL_INTERVAL_SECONDS_ENV = "FASTQSIM_POLL_INTERVAL_SECONDS"
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
_TERMINAL_FAILED = {
JobStatus.FAILED,
JobStatus.CANCELLED,
}
def _normalize_status(raw: str) -> str:
status = (raw or "").strip().lower()
if status == "canceled":
return "cancelled"
return status
def _extract_job_id(job_or_id: Union[str, Job]) -> str:
if isinstance(job_or_id, Job):
return job_or_id.job_id
return job_or_id
def _normalize_ids(
job_ids: Union[str, List[str], Job, List[Job]],
) -> List[str]:
if isinstance(job_ids, (str, Job)):
return [_extract_job_id(job_ids)]
return [_extract_job_id(j) for j in job_ids]
def _parse_timestamp(raw: Any) -> Optional[datetime]:
if raw is None:
return None
try:
return datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
except Exception:
return None
def _decode_complex_entry(value: Any) -> Any:
if isinstance(value, dict) and "real" in value and "imag" in value:
try:
return complex(float(value["real"]), float(value["imag"]))
except Exception:
return value
return value
def _decode_complex_vector(value: Any) -> Any:
if not isinstance(value, list):
return value
return [_decode_complex_entry(v) for v in value]
def _extract_session_id_from_pod_token(token: str) -> str:
"""Derive the session_id claim from a pod token."""
try:
payload_enc = token[4:].split(".", 1)[0]
padding = "=" * (-len(payload_enc) % 4)
payload_raw = base64.urlsafe_b64decode(payload_enc + padding)
payload = json.loads(payload_raw.decode("utf-8"))
except Exception as exc:
raise AuthenticationError("Malformed pod token") from exc
session_id = str(payload.get("sid", "")).strip()
if not session_id:
raise AuthenticationError("Pod token is missing a session id claim")
return session_id
# ---------------------------------------------------------------------------
# Client
# ---------------------------------------------------------------------------
[docs]
class FastQSimClient:
"""
Primary interface for submitting and managing quantum simulation jobs
on the FastQSim platform.
Instantiate via fastqsim.init() or directly:
client = FastQSimClient(endpoint="https://api.qsim.io")
"""
_DEFAULT_ENDPOINT = "https://api.qsim.io"
def __init__(
self,
endpoint: Optional[str] = None,
token: Optional[str] = None,
execution_mode: Optional[str] = None,
timeout: int = 300,
max_retries: int = 3,
) -> None:
"""
Parameters:
endpoint: Base API URL. Falls back to FASTQUBIT_ENDPOINT env var,
then the default production URL.
token: Pod-scoped session token (fqp_... format). Falls back to
FASTQUBIT_SESSION_TOKEN env var.
execution_mode: 'cloud' or 'local'. Falls back to
FASTQSIM_EXECUTION_MODE env var, then 'cloud'.
timeout: Per-request timeout in seconds.
max_retries: Number of automatic retries on transient failures
(429, 500, 502, 503, 504).
Auth resolution order (kwargs win over env):
token
FASTQUBIT_SESSION_TOKEN
FASTQUBIT_ENDPOINT
FASTQSIM_EXECUTION_MODE
"""
self._endpoint = (
endpoint
or os.environ.get("FASTQUBIT_ENDPOINT", self._DEFAULT_ENDPOINT)
).rstrip("/")
# Only pod-scoped tokens are allowed in the SDK runtime.
resolved_token = (token or "").strip() or os.environ.get("FASTQUBIT_SESSION_TOKEN", "").strip()
if not resolved_token:
raise AuthenticationError(
"No pod token available. Set FASTQUBIT_SESSION_TOKEN "
"or pass token='fqp_...'."
)
if not resolved_token.startswith("fqp_"):
raise AuthenticationError(
"FastQSim SDK only accepts pod-scoped tokens (fqp_...). "
"Integrator keys (fqi_...) are not supported in SDK runtime."
)
self._api_key = resolved_token
self._session_id = _extract_session_id_from_pod_token(resolved_token)
self._is_pod_auth = True
self._execution_mode = (
execution_mode
or os.environ.get("FASTQSIM_EXECUTION_MODE", "cloud")
)
self._timeout = timeout
self._default_poll_interval_seconds = self._resolve_default_poll_interval_seconds()
# Configure requests Session with retry logic
self._session = requests.Session()
retry = Retry(
total=max_retries, # Maximum total retry attempts
backoff_factor=0.5, # Wait: 0.5s, 1s, 2s, 4s... between retries
status_forcelist=[429, 500, 502, 503, 504], # Retry on these HTTP codes
allowed_methods=["GET", "POST", "DELETE"], # Only retry these methods
raise_on_status=False, # Don't raise — let _raise_for_status() handle it
)
adapter = HTTPAdapter(max_retries=retry)
self._session.mount("https://", adapter)
self._session.mount("http://", adapter)
headers: Dict[str, str] = {
"Content-Type": "application/json",
"Accept": "application/json",
}
headers["Authorization"] = f"Bearer {self._api_key}"
self._session.headers.update(headers)
# Validate connectivity on construction
self._validate_connection()
# ------------------------------------------------------------------
# Public: job submission
# ------------------------------------------------------------------
[docs]
def run(
self,
circuit: Union[str, List[str]],
backend: str = "qiskit",
device: str = "cpu",
shots: int = 1024,
simulation_type: Optional[str] = None,
seed: Optional[int] = None,
metadata: Optional[Dict[str, Any]] = None,
asynchronous: bool = False,
job_name: Optional[str] = None,
options: Optional[Dict[str, Any]] = None,
tags: Optional[Dict[str, Any]] = None,
) -> Union[Job, List[Job]]:
"""
Submit one or more OpenQASM 3.0 circuits for simulation.
Args:
circuit: OpenQASM 3.0 string OR list of strings.
Native circuit objects are NOT accepted.
backend: 'qiskit', 'cirq', or 'pennylane'.
device: 'cpu', 'gpu', or 'quantum'.
shots: Number of measurement repetitions.
simulation_type: Simulation method to use. One of:
- 'statevector' — full state-vector (default, all gates).
- 'density_matrix' — open-system / noisy sim (all gates).
- 'mps' — matrix product state (low-entanglement circuits).
- 'chp' — stabilizer/Clifford (Clifford gates only).
If None, the backend selects its default (statevector).
seed: RNG seed for reproducibility. Supported by all backends.
When set, the same circuit produces identical counts on
every run. When None, a fresh random seed is used.
metadata: Arbitrary key-value metadata stored with the job.
asynchronous: Return immediately (True) or block until done (False).
job_name: Human-readable label for the job.
options: Backend/device-specific options dict.
tags: Key-value tags attached to the job.
Returns:
A single Job (or list of Jobs if circuit was a list).
"""
if simulation_type is not None and simulation_type not in _VALID_SIMULATION_TYPES:
raise ValidationError(
f"Invalid simulation_type '{simulation_type}'. "
f"Must be one of: {sorted(_VALID_SIMULATION_TYPES)}"
)
if isinstance(circuit, list):
return self.run_batch(
circuits=circuit,
backend=backend,
device=device,
shots=shots,
simulation_type=simulation_type,
seed=seed,
metadata=metadata,
job_name=job_name,
options=options,
tags=tags,
)
if not isinstance(circuit, str):
raise ValidationError(
"circuit must be an OpenQASM 3.0 string. "
"Native circuit objects are not accepted by the SDK."
)
return self._submit_single_job(
circuit=circuit,
session_id=self._session_id,
backend=backend,
device=device,
shots=shots,
simulation_type=simulation_type,
seed=seed,
metadata=metadata,
asynchronous=asynchronous,
job_name=job_name,
options=options,
tags=tags,
)
[docs]
def run_batch(
self,
circuits: List[str],
backend: str = "qiskit",
device: str = "cpu",
shots: int = 1024,
max_parallel: int = 10,
simulation_type: Optional[str] = None,
seed: Optional[int] = None,
metadata: Optional[Dict[str, Any]] = None,
job_name: Optional[str] = None,
options: Optional[Dict[str, Any]] = None,
tags: Optional[Dict[str, Any]] = None,
) -> List[Job]:
"""
Submit multiple OpenQASM 3.0 circuits as a batch.
All circuits share the same backend, device, shots, simulation_type,
and seed. Batch jobs are always asynchronous — the server queues them
and returns Job objects immediately. Use client.wait() to block.
Args:
circuits: List of OpenQASM 3.0 strings.
backend: Target backend for all circuits.
device: Target device for all circuits.
shots: Shots per circuit.
max_parallel: Maximum circuits submitted concurrently (client-side threads).
simulation_type: Simulation method ('statevector', 'density_matrix',
'mps', 'chp'). Applied to every circuit in the batch.
seed: RNG seed applied to every circuit.
metadata: Key-value metadata stored with every job in the batch.
job_name: Label prefix applied to all jobs in the batch.
options: Backend/device-specific options applied to all circuits.
tags: Key-value tags attached to all jobs in the batch.
Returns:
List of Job objects (one per circuit, preserving input order).
"""
if simulation_type is not None and simulation_type not in _VALID_SIMULATION_TYPES:
raise ValidationError(
f"Invalid simulation_type '{simulation_type}'. "
f"Must be one of: {sorted(_VALID_SIMULATION_TYPES)}"
)
if not circuits:
return []
for circuit in circuits:
if not isinstance(circuit, str):
raise ValidationError(
"Each circuit in circuits must be an OpenQASM 3.0 string."
)
def _submit_one(circuit: str) -> Job:
return self._submit_single_job(
circuit=circuit,
session_id=self._session_id,
backend=backend,
device=device,
shots=shots,
simulation_type=simulation_type,
seed=seed,
metadata=metadata,
asynchronous=True,
job_name=job_name,
options=options,
tags=tags,
)
workers = min(max_parallel, len(circuits))
with ThreadPoolExecutor(max_workers=workers) as executor:
jobs = list(executor.map(_submit_one, circuits))
return jobs
def _submit_single_job(
self,
*,
circuit: str,
session_id: str,
backend: str,
device: str,
shots: int,
simulation_type: Optional[str],
seed: Optional[int],
metadata: Optional[Dict[str, Any]],
asynchronous: bool,
job_name: Optional[str],
options: Optional[Dict[str, Any]],
tags: Optional[Dict[str, Any]],
) -> Job:
payload: Dict[str, Any] = {
"session_id": session_id,
"circuit_qasm": circuit,
"backend": backend,
"shots": shots,
"device": device,
"simulation_type": simulation_type or "statevector",
"metadata": metadata or {},
"options": options or {},
"tags": tags or {},
}
if seed is not None:
payload["seed"] = seed
if job_name:
payload["job_name"] = job_name
data = self._post("/jobs/submit", payload)
job = Job(data, client=self)
if not asynchronous:
job._data = self._poll_until_terminal(
job.job_id,
poll_interval=self._default_poll_interval_seconds,
timeout=self._timeout,
)
return job
# ------------------------------------------------------------------
# Public: job retrieval
# ------------------------------------------------------------------
[docs]
def get(
self,
job_ids: Union[str, List[str], "Job", List["Job"]],
) -> Union[Job, List[Job]]:
"""
Fetch the latest metadata for one or more jobs.
Returns a single Job when given a single ID, list otherwise.
"""
ids = _normalize_ids(job_ids)
if len(ids) == 1:
return Job(self._fetch_job(ids[0]), client=self)
return [Job(self._fetch_job(jid), client=self) for jid in ids]
[docs]
def wait(
self,
job_ids: Union[str, List[str], "Job", List["Job"]],
timeout: Optional[int] = None,
poll_interval: Optional[float] = None,
) -> Union[Job, List[Job]]:
"""
Block until one or more jobs reach a terminal state.
Args:
job_ids: Job ID(s) or Job object(s).
timeout: Maximum seconds to wait. Raises JobTimeoutError if exceeded.
poll_interval: Seconds between status checks. When None,
uses FASTQSIM_POLL_INTERVAL_SECONDS (default 1.0).
Returns:
Updated Job(s) in terminal state.
"""
ids = _normalize_ids(job_ids)
resolved_poll_interval = (
self._default_poll_interval_seconds
if poll_interval is None
else poll_interval
)
results = []
for jid in ids:
terminal_data = self._poll_until_terminal(
jid,
poll_interval=resolved_poll_interval,
timeout=timeout,
)
results.append(Job(terminal_data, client=self))
return results[0] if len(results) == 1 else results
# ------------------------------------------------------------------
# Public: job cancellation
# ------------------------------------------------------------------
[docs]
def cancel(
self,
job_ids: Union[str, List[str], "Job", List["Job"]],
) -> Union[Job, List[Job]]:
"""
Request cancellation of one or more jobs.
Returns updated Job object(s). Raises JobCouldNotCancelError if
the server rejects the cancellation.
"""
ids = _normalize_ids(job_ids)
results = [Job(self._cancel_job_raw(jid), client=self) for jid in ids]
return results[0] if len(results) == 1 else results
# ------------------------------------------------------------------
# Public: job search
# ------------------------------------------------------------------
[docs]
def search(
self,
run_status: Optional[str] = None,
created_later_than: Optional[Union[str, datetime]] = None,
limit: int = 50,
backend: Optional[str] = None,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None,
) -> List[Job]:
"""
Search jobs with optional filters.
Args:
run_status: Filter by status string (e.g. 'completed', 'failed').
created_later_than: Only return jobs created after this datetime.
limit: Maximum number of results to return (1–500).
backend: Filter by backend name ('qiskit', 'cirq', 'pennylane').
start_date: Filter by creation start date (inclusive).
end_date: Filter by creation end date (inclusive).
Returns:
List of matching Job objects, most-recently-updated first.
"""
params: Dict[str, Any] = {"limit": limit}
if run_status:
params["status"] = run_status
if backend:
params["backend"] = backend
if created_later_than:
ts = (
created_later_than.isoformat()
if isinstance(created_later_than, datetime)
else created_later_than
)
params["created_later_than"] = ts
if start_date:
params["start_date"] = start_date.isoformat()
if end_date:
params["end_date"] = end_date.isoformat()
data = self._get("/jobs", params=params)
return [Job(j, client=self) for j in data.get("items", [])]
# ------------------------------------------------------------------
# Public: session
# ------------------------------------------------------------------
[docs]
def get_session(self) -> Session:
"""Session inspection is not available in pod-token SDK mode."""
raise ValidationError(
"Session inspection is not available in FastQSim SDK pod-token mode."
)
[docs]
def start_session(self) -> Session:
"""Session creation is not available in pod-token SDK mode."""
raise ValidationError(
"Session creation is not available in FastQSim SDK pod-token mode. "
"Sessions are provisioned by the integrator control plane via /sessions/start."
)
# ------------------------------------------------------------------
# Internal: polling helpers (called by Job methods in types.py)
# ------------------------------------------------------------------
def _poll_until_terminal(
self,
job_id: str,
poll_interval: float,
timeout: Optional[int] = None,
) -> Dict[str, Any]:
"""Block until job reaches a terminal state or timeout is exceeded.
Returns the final job data dict so callers avoid a redundant fetch.
"""
if poll_interval <= 0:
raise ValidationError("poll_interval must be > 0")
deadline = (time.monotonic() + timeout) if timeout else None
while True:
data = self._fetch_job(job_id)
status = JobStatus(_normalize_status(data.get("status", "queued")))
if status.is_terminal:
return data
if deadline and time.monotonic() >= deadline:
raise JobTimeoutError(job_id, timeout) # type: ignore[arg-type]
time.sleep(poll_interval)
def _wait_and_get_result(
self, job_id: str, timeout: Optional[int] = None
) -> Result:
"""Poll until terminal, then raise or return Result."""
data = self._poll_until_terminal(
job_id,
poll_interval=self._default_poll_interval_seconds,
timeout=timeout,
)
status = JobStatus(_normalize_status(data.get("status", "queued")))
if status in _TERMINAL_FAILED:
raise JobFailedError(
job_id, reason=data.get("error_message", "")
)
counts = data.get("result_counts") or {}
metadata = data.get("result_metadata") or {}
result_payload: Dict[str, Any] = {
"counts": counts,
"metadata": metadata,
}
if "probabilities" in metadata:
result_payload["probabilities"] = metadata.get("probabilities")
elif "probs" in metadata:
result_payload["probabilities"] = metadata.get("probs")
if "statevector" in metadata:
result_payload["statevector"] = _decode_complex_vector(metadata.get("statevector"))
if "density_matrix" in metadata:
result_payload["density_matrix"] = metadata.get("density_matrix")
if data.get("has_external_artifact"):
artifact = self._fetch_external_artifact(job_id)
if isinstance(artifact, dict):
if "counts" in artifact and not result_payload["counts"]:
result_payload["counts"] = artifact.get("counts") or {}
if "metadata" in artifact and isinstance(artifact.get("metadata"), dict):
merged_meta = dict(result_payload["metadata"])
merged_meta.update(artifact.get("metadata") or {})
result_payload["metadata"] = merged_meta
if "statevector" in artifact:
result_payload["statevector"] = _decode_complex_vector(artifact.get("statevector"))
if "density_matrix" in artifact:
result_payload["density_matrix"] = artifact.get("density_matrix")
if "probabilities" in artifact:
result_payload["probabilities"] = artifact.get("probabilities")
elif "probs" in artifact:
result_payload["probabilities"] = artifact.get("probs")
started_at = (
_parse_timestamp(data.get("execution_started_at"))
or _parse_timestamp((result_payload["metadata"] or {}).get("execution_started_at"))
)
finished_at = (
_parse_timestamp(data.get("execution_finished_at"))
or _parse_timestamp((result_payload["metadata"] or {}).get("execution_finished_at"))
)
if started_at is not None and finished_at is not None:
duration = (finished_at - started_at).total_seconds()
if duration >= 0:
result_payload["execution_time"] = duration
return Result(result_payload)
def _get_counts(self, job_id: str) -> Dict[str, int]:
data = self._fetch_job(job_id)
status = JobStatus(_normalize_status(data.get("status", "queued")))
if not status.is_terminal:
raise JobNotCompleteError(job_id, current_status=status.value)
if status != JobStatus.COMPLETED:
raise JobFailedError(job_id, reason=data.get("error_message", ""))
counts = data.get("result_counts")
if counts is None:
raise JobDoesNotContainCountsError(
f"Job '{job_id}' does not contain measurement counts."
)
return counts
def _get_statevector(self, job_id: str) -> List[complex]:
data = self._fetch_job(job_id)
status = JobStatus(_normalize_status(data.get("status", "queued")))
if not status.is_terminal:
raise JobNotCompleteError(job_id, current_status=status.value)
if status != JobStatus.COMPLETED:
raise JobFailedError(job_id, reason=data.get("error_message", ""))
metadata = data.get("result_metadata") or {}
if "statevector" in metadata:
statevector = _decode_complex_vector(metadata["statevector"])
if isinstance(statevector, list):
return statevector
if data.get("has_external_artifact"):
artifact = self._fetch_external_artifact(job_id)
if isinstance(artifact, dict) and "statevector" in artifact:
statevector = _decode_complex_vector(artifact["statevector"])
if isinstance(statevector, list):
return statevector
raise JobStatevectorTooLargeError(
f"Statevector for job '{job_id}' is not available in current FastQubit responses."
)
def _cancel_job(self, job_id: str) -> bool:
result = self._cancel_job_raw(job_id)
return JobStatus(_normalize_status(result.get("status", ""))) == JobStatus.CANCELLED
def _cancel_job_raw(self, job_id: str) -> Dict[str, Any]:
response = self._session.post(
f"{self._endpoint}/jobs/{job_id}/cancel",
timeout=self._timeout,
)
if response.status_code == 409:
raise JobCouldNotCancelError(
job_id, reason=response.json().get("detail", "")
)
self._raise_for_status(response)
return response.json()
def _fetch_job(self, job_id: str, wait: int = 0) -> Dict[str, Any]:
params = {"wait": wait} if wait > 0 else None
response = self._session.get(
f"{self._endpoint}/jobs/{job_id}",
params=params,
timeout=max(self._timeout, wait + 10),
)
if response.status_code == 404:
raise JobNotFoundError(job_id)
self._raise_for_status(response)
return response.json()
def _refresh_token(self, session_id: str) -> None:
raise ValidationError(
"WebSocket token refresh is not available in FastQSim SDK pod-token mode."
)
def _close_session(self, session_id: str) -> None:
raise ValidationError(
"Session termination is not available in FastQSim SDK pod-token mode."
)
# ------------------------------------------------------------------
# Internal: HTTP primitives
# ------------------------------------------------------------------
def _get(
self, path: str, params: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
try:
response = self._session.get(
f"{self._endpoint}{path}",
params=params,
timeout=self._timeout,
)
except requests.exceptions.ConnectionError as exc:
raise FastQSimConnectionError(
f"Cannot reach FastQSim API at {self._endpoint}: {exc}"
) from exc
self._raise_for_status(response)
return response.json()
def _post(self, path: str, payload: Dict[str, Any]) -> Dict[str, Any]:
try:
response = self._session.post(
f"{self._endpoint}{path}",
json=payload,
timeout=self._timeout,
)
except requests.exceptions.ConnectionError as exc:
raise FastQSimConnectionError(
f"Cannot reach FastQSim API at {self._endpoint}: {exc}"
) from exc
self._raise_for_status(response)
return response.json()
def _fetch_external_artifact(self, job_id: str) -> Dict[str, Any]:
url_meta = self._get(f"/jobs/{job_id}/result-url")
if not url_meta.get("available"):
return {}
result_url = str(url_meta.get("result_url") or "").strip()
if not result_url:
return {}
# Presigned S3 URLs are self-authenticated via query params.
# Drop API auth headers from the session to avoid S3 signature mismatches.
response = self._session.get(
result_url,
timeout=self._timeout,
headers={
"Authorization": None,
"X-End-User-Id": None,
"Content-Type": None,
"Accept": None,
},
)
response.raise_for_status()
payload = response.json()
if isinstance(payload, dict):
return payload
return {}
def _raise_for_status(self, response: requests.Response) -> None:
"""Map HTTP error codes to SDK exceptions."""
code = response.status_code
if code < 400:
return
try:
detail = response.json().get("detail", response.text)
except Exception:
detail = response.text
if code == 401:
raise AuthenticationError(f"Authentication failed: {detail}")
if code == 403:
raise UnauthorizedAccessError(f"Access denied: {detail}")
if code == 404:
raise JobNotFoundError(str(detail))
if code == 422:
raise ValidationError(f"Validation error: {detail}")
if code == 402:
raise QuotaExceededError(f"Quota exceeded: {detail}")
response.raise_for_status()
def _validate_connection(self) -> None:
"""Ping /health to verify endpoint reachability and auth posture."""
try:
response = self._session.get(
f"{self._endpoint}/health",
timeout=10,
)
except requests.exceptions.ConnectionError as exc:
raise FastQSimConnectionError(
f"Cannot reach FastQSim API at {self._endpoint}: {exc}"
) from exc
if response.status_code == 401:
raise AuthenticationError(
"Invalid or expired pod token. "
"Verify FASTQUBIT_SESSION_TOKEN is set correctly (format: fqp_...)."
)
if response.status_code == 403:
raise UnauthorizedAccessError(
"Pod token lacks permission to access this endpoint. "
"Check the session is active and token scope is valid."
)
def _resolve_default_poll_interval_seconds(self) -> float:
raw_value = (os.environ.get(_POLL_INTERVAL_SECONDS_ENV) or "").strip()
if not raw_value:
return _DEFAULT_POLL_INTERVAL_SECONDS
try:
parsed_value = float(raw_value)
if parsed_value <= 0:
raise ValueError
except ValueError:
warnings.warn(
(
f"Ignoring invalid {_POLL_INTERVAL_SECONDS_ENV}={raw_value!r}; "
f"using {_DEFAULT_POLL_INTERVAL_SECONDS}."
),
stacklevel=2,
)
return _DEFAULT_POLL_INTERVAL_SECONDS
return parsed_value
def __repr__(self) -> str:
return (
f"FastQSimClient(endpoint='{self._endpoint}', "
f"mode='{self._execution_mode}', auth='pod-token')"
)
# ---------------------------------------------------------------------------
# Module-level convenience (global client instance)
# ---------------------------------------------------------------------------
_global_client: Optional[FastQSimClient] = None
[docs]
def init(
endpoint: Optional[str] = None,
token: Optional[str] = None,
execution_mode: Optional[str] = None,
timeout: int = 300,
max_retries: int = 3,
) -> FastQSimClient:
"""
Initialise the global FastQSim client and return it.
The recommended usage is to call this with no arguments. The BFF or
QCanvas injects runtime environment variables into the user-session pod,
so the SDK picks them up without manual configuration:
client = fastqsim.init()
Args:
endpoint: Base API endpoint. Defaults to FASTQUBIT_ENDPOINT env var.
token: Pod-scoped session token (fqp_... format). Defaults to
FASTQUBIT_SESSION_TOKEN env var.
execution_mode: 'cloud' or 'local'. Defaults to
FASTQSIM_EXECUTION_MODE env var or 'cloud'.
timeout: Per-request timeout in seconds.
max_retries: Retry attempts for transient HTTP failures.
Returns:
FastQSimClient instance (also stored as the global client).
Raises:
AuthenticationError: No valid auth credential available.
FastQSimConnectionError: Cannot reach the API endpoint.
"""
global _global_client
if _global_client is not None:
warnings.warn(
"fastqsim.init() called more than once. The previous client is being "
"replaced. If this is intentional, call fastqsim.reset() first to "
"suppress this warning.",
stacklevel=2,
)
_global_client = FastQSimClient(
endpoint=endpoint,
token=token,
execution_mode=execution_mode,
timeout=timeout,
max_retries=max_retries,
)
return _global_client
def _get_global_client() -> FastQSimClient:
if _global_client is None:
raise RuntimeError(
"FastQSim client not initialised. Call fastqsim.init() first."
)
return _global_client
[docs]
def reset() -> None:
"""
Clear the global client instance.
Call this before fastqsim.init() if you intentionally want to
re-initialise with a different token or config without a warning.
fastqsim.reset()
# Update environment first, then init again.
fastqsim.init()
"""
global _global_client
_global_client = None