uniqc.gateway.api.tasks module

Task listing and management API — /api/tasks.

class uniqc.gateway.api.tasks.TaskIdsRequest(*, task_ids=<factory>)[source]

Bases: BaseModel

Parameters:

task_ids (list[str])

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

task_ids: list[str]
uniqc.gateway.api.tasks.archive_expired_tasks(hours=72, terminal_only=True)[source]

Archive active tasks older than hours.

By default only terminal tasks are archived, so pending/running jobs are not hidden from the active task view.

Parameters:
Return type:

dict[str, Any]

uniqc.gateway.api.tasks.bulk_archive_tasks(payload)[source]

Move active tasks into the archive by id.

Parameters:

payload (TaskIdsRequest)

Return type:

dict[str, Any]

uniqc.gateway.api.tasks.bulk_delete_tasks(payload)[source]

Permanently delete active tasks by id.

Parameters:

payload (TaskIdsRequest)

Return type:

dict[str, Any]

uniqc.gateway.api.tasks.delete_task(task_id)[source]

Permanently delete a task.

Parameters:

task_id (str)

Return type:

dict[str, str]

uniqc.gateway.api.tasks.get_task(task_id)[source]

Return full task details including result.

Parameters:

task_id (str)

Return type:

dict[str, Any]

uniqc.gateway.api.tasks.get_task_shards(task_id)[source]

Return the shard mapping for a uniqc task id.

Each shard records a single submission to the underlying cloud platform; circuit_count indicates how many circuits the shard covers (>= 1 for native-batch platforms, == 1 otherwise). Result payloads are excluded — fetch GET /api/tasks/{task_id} for the aggregated result.

Parameters:

task_id (str)

Return type:

dict[str, Any]

uniqc.gateway.api.tasks.list_tasks(status=None, backend=None, limit=100, offset=0)[source]

List active (non-archived) tasks.

Query params:

status: filter by task status (pending/running/success/failed/cancelled) backend: filter by backend name limit: max rows returned (default 100) offset: number of matching rows to skip

Parameters:
  • status (str | None)

  • backend (str | None)

  • limit (int | None)

  • offset (int | None)

Return type:

dict[str, Any]

uniqc.gateway.api.tasks.task_counts()[source]

Return count of active tasks grouped by status.

Return type:

dict[str, int]