uniqc.task.store module

SQLite-backed task storage for UnifiedQuantum.

This module is the single source of truth for task persistence. Both uniqc.task_manager and uniqc.task.persistence delegate to TaskStore; there is no JSON/JSONL fallback path.

Storage layout:

~/.uniqc/cache/
    tasks.sqlite          # single database file

Schema versioning (best practices)

The database uses SQLite’s built-in metadata slots in the file header rather than a custom bookkeeping table:

• PRAGMA application_id holds APPLICATION_ID (the 4 bytes "UNIC"). On open we refuse to touch a file whose application_id is set to something else; a zero value means a fresh DB we can stamp.

• PRAGMA user_version holds the integer schema version. It is bumped by each migration and used to decide what else to run.

MIGRATIONS is an ordered list of (target_version, migrate_fn) tuples. To evolve the schema:

1. Append a new tuple (N, _apply_vN) where N = CURRENT_SCHEMA_VERSION + 1.

2. Bump CURRENT_SCHEMA_VERSION to N.

3. Implement _apply_vN(conn) (DDL or DML on the given connection).

On open, TaskStore iterates the list and runs each outstanding migration in its own transaction; user_version is stamped as part of the same transaction so a crashed migration rolls back the version bump too.

class uniqc.task.store.TaskInfo(task_id, backend, status=TaskStatus.PENDING, result=None, shots=1000, submit_time=<factory>, update_time=<factory>, metadata=<factory>)

Bases: object

Information about a submitted task.

Variables:

• task_id (str) – Unique identifier for the task.

• backend (str) – The backend where the task was submitted.

• status (str) – Current status of the task.

• result (dict | None) – Task result (if completed).

• shots (int) – Number of shots requested.

• submit_time (str) – ISO format timestamp of submission.

• update_time (str) – ISO format timestamp of last status update.

• metadata (dict) – Additional metadata about the task.

Parameters:

• task_id (str)

• backend (str)

• status (str)

• result (dict | None)

• shots (int)

• submit_time (str)

• update_time (str)

• metadata (dict)

backend: str

classmethod from_dict(data)

Create from dictionary.

Parameters:

data (dict[str, Any])

Return type:

TaskInfo

metadata: dict

result: dict | None = None

shots: int = 1000

status: str = 'pending'

submit_time: str

task_id: str

to_dict()

Convert to dictionary.

Return type:

dict[str, Any]

update_time: str

class uniqc.task.store.TaskStatus(*values)

Bases: str, Enum

Enumeration of task statuses.

CANCELLED = 'cancelled'

FAILED = 'failed'

PENDING = 'pending'

RUNNING = 'running'

SUCCESS = 'success'

class uniqc.task.store.TaskStore(cache_dir=None)

Bases: object

SQLite-backed storage for task records.

TaskStore is safe to construct multiple times against the same cache directory; all operations use short-lived connections guarded by a per-instance lock.

Parameters:

cache_dir (Path | str | None) – Directory that holds tasks.sqlite. Defaults to ~/.uniqc/cache/. The directory is created if missing.

cache_dir: Path

clear_all()

Delete the on-disk SQLite file (and its WAL/SHM siblings).

Matches the pre-SQLite clear_cache semantics: the cache file disappears. A subsequent operation re-initialises the schema.

Return type:

None

clear_completed(terminal_statuses=('success', 'failed', 'cancelled'))

Delete tasks whose status is in terminal_statuses.

Parameters:

terminal_statuses (tuple[str, ...])

Return type:

int

count(*, status=None, backend=None)

Parameters:

• status (str | None)

• backend (str | None)

Return type:

int

db_path: Path

delete(task_id)

Remove a single task. Returns True if it existed.

Parameters:

task_id (str)

Return type:

bool

get(task_id)

Return a task by id, or None if missing.

Parameters:

task_id (str)

Return type:

TaskInfo | None

list(*, status=None, backend=None, limit=None)

List tasks, newest first (by submit_time).

Parameters:

• status (str | None)

• backend (str | None)

• limit (int | None)

Return type:

list[TaskInfo]

save(task_info)

Insert or update a task record (by task_id).

Also stamps task_info.update_time with the current timestamp so callers see the same value that was persisted.

Parameters:

task_info (TaskInfo)

Return type:

None