Queue#

class daq_queuing_service.task_queue.queue.TaskQueue(converter: Converter, broadcaster: Broadcaster[Literal['state_update', 'queue_update', 'history_update', 'tasks_update', 'call_queue_update', 'call_history_update']], runtime_estimator_client: RuntimeEstimatorClient | None = None, instrument: str | None = None)[source]#

Bases: object

__init__(converter: Converter, broadcaster: Broadcaster[Literal['state_update', 'queue_update', 'history_update', 'tasks_update', 'call_queue_update', 'call_history_update']], runtime_estimator_client: RuntimeEstimatorClient | None = None, instrument: str | None = None)[source]#

async get_next_call_once_available() → BlueapiCall[source]#

Waits until a call is available before returning the call. A call is available if it’s at the top of the call queue, is not already in progress or claimed, and the queue is not paused.

Returns:: The call at the top of the queue
Return type:: BlueapiCall

async wait_until_call_available()[source]#: Waits until a task is available before returning. A task is available if it’s at the top of the queue, is not already in progress or claimed, and the queue is not paused.

async return_call_to_queue(call: BlueapiCall)[source]#

Returns a task to the queue that had previously been claimed

Parameters:: task (Task) – The task to return
Raises:: TaskNotClaimedError – Raised if the task’s status is not ‘Claimed’

async complete_call(call: BlueapiCall, result: TaskResult)[source]#

Sets a task to complete, removes it from the queue and adds it to history

Parameters:

task (Task) – Task to be completed
result (TaskResult) – The result of the task from blueapi

async fail_call(call: BlueapiCall, errors: list[str | TaskError] | None = None)[source]#

Sets a task to failed, removes it from the call queue and adds it to history.

Parameters:

task (Task) – The task to fail
errors (list[str | TaskError] | None, optional) – A list of errors that
None. (occurred when trying to run the task. Defaults to)

async get_task_by_id(task_id: str) → TaskWithPosition[source]#

Returns a task based on it’s task ID

Parameters:: task_id (str) – Task ID of the task
Returns:: A copy of the task with a position field.
Return type:: TaskWithPosition
Raises:: TaskNotFoundError – Raised if the no task exists with the requested task ID.

async get_task_by_position(position: int) → TaskWithPosition | None[source]#

Return a task based on it’s position in the queue

Parameters:: position (int) – The position of the task to be returned.
Returns:: A copy of the task with a position field, or None if no task exists at the requested position.
Return type:: TaskWithPosition | None

async get_queue() → list[TaskWithPosition][source]#

Get the entire queue (not including history)

Returns:: A list of the tasks in the queue, in the order they will be run in.
Return type:: list[TaskWithPosition]

async get_history() → list[TaskWithPosition][source]#

Get the history list.

Returns:: A list of the tasks in the history list, in chronological order.
Return type:: list[TaskWithPosition]

async get_tasks() → list[TaskWithPosition][source]#

Get all the tasks in the queue and history list.

Returns:: A list of tasks in chronological order, starting with the history.
Return type:: list[TaskWithPosition]

async add_tasks(tasks: list[Task], position: int | None = None) → None[source]#

Add tasks to the queue. Can specify a position to add the tasks in at. This position will apply to the first task in the list, with each subsequent task having a position of 1 more than the previous. By default adds tasks to the end of the queue.

Parameters:

tasks (list[Task]) – List of tasks to add
position (int | None, optional) – Position of the tasks. Defaults to None.

async move_task(task_id: str, position: int) → int[source]#

Move a task into a different position. If the requested position is unavailable, will move to the closest available position. If the requested task cannot be moved because it is complete or in progress, an error will be raised.

Parameters:

task_id (str) – ID of the task to be moved
new_position (int) – New position of the task

Returns:

The new position of the task (may be different to what was requested)

Return type:

int

async cancel_tasks(task_ids: Sequence[str]) → list[TaskWithPosition][source]#

Remove tasks from the queue. If one or more of the requested tasks cannot be cancelled as they are complete or in progress, an error will be raised and none of the requested tasks will be cancelled.

Parameters:: task_ids (Sequence[str]) – List of task IDs to cancel
Returns:: List of the task objects that were removed from the queue.
Return type:: list[TaskWithPosition]

async cancel_all_tasks() → list[TaskWithPosition][source]#

async clear_history()[source]#: Clears the history list. Any task in the history list at the time will be deleted permanently and inaccessible.

async pause_queue(reason: PauseReason) → QueueState[source]#

Pause the queue if running.

Parameters:: paused (PauseReason) – The reason for pausing the queue.
Returns:: The new state of the queue.
Return type:: QueueState

async resume_queue() → QueueState[source]#

Resume the queue if paused.

Returns:: The new state of the queue.
Return type:: QueueState

property state#

property length#

async get_call_queue() → list[BlueapiCallResponse][source]#

async get_call_history() → list[BlueapiCallResponse][source]#