qpandalite.task_manager module#

Task management with local caching for quantum computing backends.

This module provides a unified interface for submitting quantum tasks, managing task lifecycle, and caching results locally.

Environment Variables:
QPANDALITE_DUMMY: Set to 'true', '1', or 'yes' to enable dummy mode.

When enabled, all task submissions use local simulation instead of real quantum backends. Useful for development and testing.

Usage:

from qpandalite.task_manager import submit_task, query_task, wait_for_result from qpandalite.circuit_builder import Circuit from qpandalite.backend import get_backend

# Create a circuit circuit = Circuit() circuit.h(0) circuit.cnot(0, 1) circuit.measure(0, 1)

# Submit task (use QPANDALITE_DUMMY=true for local simulation) task_id = submit_task(circuit, backend='quafu', shots=1000)

# Wait for result result = wait_for_result(task_id, backend='quafu', timeout=300)

# Query task status info = query_task(task_id, backend='quafu') print(info['status']) # 'running', 'success', or 'failed'

# Explicitly use dummy mode for a single submission task_id = submit_task(circuit, backend='quafu', dummy=True)

class qpandalite.task_manager.TaskInfo(task_id, backend, status=TaskStatus.PENDING, result=None, shots=1000, submit_time=<factory>, update_time=<factory>, metadata=<factory>)[源代码]

基类：object

Information about a submitted task.

参数:

  • task_id (str)

  • backend (str)

  • status (str)

  • result (dict | None)

  • shots (int)

  • submit_time (str)

  • update_time (str)

  • metadata (dict)

task_id

Unique identifier for the task.

Type:

str

backend

The backend where the task was submitted.

Type:

str

status

Current status of the task.

Type:

str

result

Task result (if completed).

Type:

dict | None

shots

Number of shots requested.

Type:

int

submit_time

ISO format timestamp of submission.

Type:

str

update_time

ISO format timestamp of last status update.

Type:

str

metadata

Additional metadata about the task.

Type:

dict

backend: str
classmethod from_dict(data)[源代码]

Create from dictionary.

参数:

data (dict[str, Any])

返回类型:

TaskInfo

metadata: dict
result: dict | None = None
shots: int = 1000
status: str = 'pending'
submit_time: str
task_id: str
to_dict()[源代码]

Convert to dictionary.

返回类型:

dict[str, Any]

update_time: str
class qpandalite.task_manager.TaskManager(cache_dir=None)[源代码]

基类：object

High-level task manager for quantum computing workflows.

This class provides a convenient interface for managing quantum tasks with persistent caching and batch operations.

示例

>>> manager = TaskManager()
>>> task_id = manager.submit(circuit, backend='quafu', shots=1000)
>>> result = manager.wait_for_result(task_id)
>>> print(result)
参数:

cache_dir (Path | str | None)

clear_cache()[源代码]

Clear all tasks from cache.

返回类型:

None

clear_completed()[源代码]

Clear completed tasks from cache.

返回:

Number of tasks removed.

返回类型:

int

list_tasks(status=None, backend=None)[源代码]

List tasks from cache.

参数:

  • status (str | None) -- Filter by status.

  • backend (str | None) -- Filter by backend.

返回:

List of matching tasks.

返回类型:

list[TaskInfo]

query(task_id, backend=None)[源代码]

Query a task's status.

参数:

  • task_id (str) -- The task ID.

  • backend (str | None) -- Optional backend name.

返回:

TaskInfo with current status.

返回类型:

TaskInfo

submit(circuit, backend, shots=1000, metadata=None, **kwargs)[源代码]

Submit a single circuit.

参数:

  • circuit (Circuit) -- The circuit to submit.

  • backend (str) -- The backend name.

  • shots (int) -- Number of shots.

  • metadata (dict | None) -- Optional metadata.

  • **kwargs (Any) -- Backend-specific parameters.

返回:

The task ID.

返回类型:

str

submit_batch(circuits, backend, shots=1000, **kwargs)[源代码]

Submit multiple circuits as a batch.

参数:

  • circuits (list[Circuit]) -- List of circuits to submit.

  • backend (str) -- The backend name.

  • shots (int) -- Number of shots per circuit.

  • **kwargs (Any) -- Backend-specific parameters.

返回:

List of task IDs.

返回类型:

list[str]

wait_for_result(task_id, backend=None, timeout=300.0, poll_interval=5.0, raise_on_failure=True)[源代码]

Wait for a task to complete.

参数:

  • task_id (str) -- The task ID.

  • backend (str | None) -- Optional backend name.

  • timeout (float) -- Maximum wait time in seconds.

  • poll_interval (float) -- Time between status checks.

  • raise_on_failure (bool) -- Whether to raise on task failure.

返回:

Task result if successful, None if timed out.

返回类型:

dict | None

qpandalite.task_manager.clear_cache(cache_dir=None)[源代码]

Clear all tasks from the cache.

参数:

cache_dir (Path | None) -- Optional custom cache directory.

返回类型:

None

qpandalite.task_manager.clear_completed_tasks(cache_dir=None)[源代码]

Remove completed tasks from the cache.

参数:

cache_dir (Path | None) -- Optional custom cache directory.

返回:

Number of tasks removed.

返回类型:

int

qpandalite.task_manager.get_task(task_id, cache_dir=None)[源代码]

Get a task from the local cache.

参数:

  • task_id (str) -- The task identifier.

  • cache_dir (Path | None) -- Optional custom cache directory.

返回:

TaskInfo if found, None otherwise.

返回类型:

TaskInfo | None

qpandalite.task_manager.is_dummy_mode()[源代码]

Check if dummy mode is enabled via environment variable.

返回:

True if QPANDALITE_DUMMY is set to 'true', '1', or 'yes'.

返回类型:

bool

qpandalite.task_manager.list_tasks(status=None, backend=None, cache_dir=None)[源代码]

List tasks from the local cache.

参数:

  • status (str | None) -- Filter by status (optional).

  • backend (str | None) -- Filter by backend (optional).

  • cache_dir (Path | None) -- Optional custom cache directory.

返回:

List of TaskInfo objects matching the filters.

返回类型:

list[TaskInfo]

qpandalite.task_manager.query_task(task_id, backend=None)[源代码]

Query the status of a task.

This function queries the backend for the current status of a task and updates the local cache.

参数:

  • task_id (str) -- The task identifier.

  • backend (str | None) -- The backend name. If None, attempts to look up from cache. Prefer using None to let the system auto-detect the correct backend.

返回:

TaskInfo with current status and result if available.

抛出:

  • TaskNotFoundError -- If the task is not found locally or remotely.

  • BackendNotFoundError -- If the backend is not recognized.

  • NetworkError -- If a network error occurs.

返回类型:

TaskInfo

示例

>>> info = query_task('task-123', backend='quafu')
>>> print(info.status)
'success'
qpandalite.task_manager.save_task(task_info, cache_dir=None)[源代码]

Save a task to the local cache.

参数:

  • task_info (TaskInfo) -- Task information to save.

  • cache_dir (Path | None) -- Optional custom cache directory.

返回类型:

None

qpandalite.task_manager.submit_batch(circuits, backend, shots=1000, dummy=None, **kwargs)[源代码]

Submit multiple circuits as a batch to a quantum backend.

参数:

  • circuits (list[Circuit]) -- List of QPanda-lite Circuits to submit.

  • backend (str) -- The backend name.

  • shots (int) -- Number of measurement shots per circuit.

  • dummy (bool | None) -- Override dummy mode. If None, uses QPANDALITE_DUMMY env var.

  • **kwargs (Any) -- Additional backend-specific parameters. - For Quafu: chip_id, auto_mapping, group_name - For OriginQ: backend_name (e.g., 'origin:wuyuan:d5'), circuit_optimize

返回:

List of task IDs assigned by the backend.

抛出:

  • BackendNotFoundError -- If the backend is not recognized.

  • BackendNotAvailableError -- If the backend is not available.

  • AuthenticationError -- If authentication fails.

  • InsufficientCreditsError -- If account has insufficient credits.

  • QuotaExceededError -- If usage quota is exceeded.

  • NetworkError -- If a network error occurs.

返回类型:

list[str]

示例

>>> circuits = [circuit1, circuit2, circuit3]
>>> task_ids = submit_batch(circuits, backend='quafu', shots=1000, chip_id='ScQ-P10')
qpandalite.task_manager.submit_task(circuit, backend, shots=1000, metadata=None, dummy=None, **kwargs)[源代码]

Submit a single circuit to a quantum backend.

This function converts the circuit to the backend's native format, submits it, and caches the task information locally.

参数:

  • circuit (Circuit) -- The QPanda-lite Circuit to submit.

  • backend (str) -- The backend name (e.g., 'originq', 'quafu', 'ibm').

  • shots (int) -- Number of measurement shots.

  • metadata (dict | None) -- Optional metadata to store with the task.

  • dummy (bool | None) -- Override dummy mode. If None, uses QPANDALITE_DUMMY env var. When True, uses local simulation instead of real backend.

  • **kwargs (Any) -- Additional backend-specific parameters. - For Quafu: chip_id, auto_mapping - For OriginQ: backend_name (e.g., 'origin:wuyuan:d5'), circuit_optimize, measurement_amend

返回:

The task ID assigned by the backend.

抛出:

  • BackendNotFoundError -- If the backend is not recognized.

  • BackendNotAvailableError -- If the backend is not available.

  • AuthenticationError -- If authentication fails.

  • InsufficientCreditsError -- If account has insufficient credits.

  • QuotaExceededError -- If usage quota is exceeded.

  • NetworkError -- If a network error occurs.

返回类型:

str

示例

>>> circuit = Circuit()
>>> circuit.h(0)
>>> circuit.measure(0)
>>> task_id = submit_task(circuit, backend='originq', shots=1000, backend_name='origin:wuyuan:d5')
>>> # Use dummy mode for local simulation
>>> task_id = submit_task(circuit, backend='quafu', dummy=True)
qpandalite.task_manager.wait_for_result(task_id, backend=None, timeout=300.0, poll_interval=5.0, raise_on_failure=True)[源代码]

Wait for a task to complete and return its result.

This function polls the task status until it completes, fails, or the timeout is reached.

参数:

  • task_id (str) -- The task identifier.

  • backend (str | None) -- The backend name. If None, attempts to look up from cache.

  • timeout (float) -- Maximum time to wait in seconds.

  • poll_interval (float) -- Time between status checks in seconds.

  • raise_on_failure (bool) -- If True, raises TaskFailedError on task failure.

返回:

The task result dictionary if successful, None if timed out.

抛出:

  • TaskTimeoutError -- If the timeout is reached before completion.

  • TaskFailedError -- If the task fails and raise_on_failure is True.

  • TaskNotFoundError -- If the task is not found.

  • NetworkError -- If a network error occurs.

返回类型:

dict | None

示例

>>> result = wait_for_result('task-123', backend='quafu', timeout=300)
>>> print(result['counts'])
{'00': 512, '11': 488}