uniqc.task.persistence module#

Task persistence (legacy dict-based API) backed by the SQLite TaskStore.

Historically this module offered a JSONL-based TaskPersistence class. All storage has been unified onto uniqc.task.store.TaskStore (SQLite); this module now provides a thin compatibility layer that keeps the same flat-dict interface (platform/status/result + extra keyword metadata).

Usage:

from uniqc.task.persistence import TaskPersistence

persistence = TaskPersistence()

# Save a task
persistence.save(
    task_id="task-123",
    platform="originq",
    status="success",
    result={"counts": {"00": 512, "11": 488}},
    shots=1000,
)

# Load a task
record = persistence.load("task-123")

# List all tasks for a platform
tasks = persistence.list_all(platform="originq")
class uniqc.task.persistence.TaskPersistence(cache_dir=None)[source]#

Bases: object

Dict-shaped task store backed by SQLite.

The dict schema exposed to callers uses platform as the field name for the backend (kept for backward compatibility with pre-unification callers). Internally, records are stored in the shared SQLite database managed by TaskStore.

Variables:

  • cache_dir – Directory containing tasks.sqlite.

  • tasks_file – Path to the SQLite database (legacy name preserved for callers that introspect the storage file).

Parameters:

cache_dir (Optional[Path])

Example

>>> persistence = TaskPersistence()
>>> persistence.save("task-1", "originq", "running")
>>> record = persistence.load("task-1")
>>> print(record['status'])
'running'
clear_completed()[source]#

Remove records whose status is terminal. Returns count removed.

Return type:

int

count(platform=None, status=None)[source]#

Count records with optional filters.

Parameters:

  • platform (str | None)

  • status (str | None)

Return type:

int

delete(task_id)[source]#

Delete a record by id. Returns True if it existed.

Parameters:

task_id (str)

Return type:

bool

list_all(platform=None, status=None, limit=None)[source]#

List records, newest first.

Parameters:

  • platform (str | None) – Filter by platform / backend name.

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

  • limit (int | None) – Max number of records.

Return type:

List[Dict[str, Any]]

list_by_platform(platform)[source]#

All records for a given platform.

Parameters:

platform (str)

Return type:

List[Dict[str, Any]]

list_pending()[source]#

Records currently in-flight (‘pending’ or ‘running’).

Return type:

List[Dict[str, Any]]

load(task_id)[source]#

Load a record by task id.

Parameters:

task_id (str)

Return type:

Dict[str, Any] | None

save(task_id, platform, status, result=None, **metadata)[source]#

Save (upsert) a task record.

Parameters:

  • task_id (str) – Unique task identifier.

  • platform (str) – Platform / backend name.

  • status (str) – Task status (‘pending’, ‘running’, ‘success’, ‘failed’).

  • result (Dict[str, Any] | None) – Optional result dict.

  • **metadata (Any) – Extra fields. shots, submit_time, update_time are recognised and promoted onto TaskInfo; anything else is stored in TaskInfo.metadata.

Return type:

None

update(task_id, **updates)[source]#

Update an existing record. update_time is refreshed.

Returns True if the record existed and was updated.

Parameters:
Return type:

bool

upsert(task_id, platform, status, result=None, **metadata)[source]#

Update if present, otherwise insert.

Parameters:
Return type:

None