From f3b718e62b8f5751221c854cefc3eddeb94bd49e Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Fri, 30 Oct 2026 16:27:19 +0000 Subject: [PATCH 01/11] Rename types --- docs/explanations/converters.md | 16 ++--- docs/reference/rest_api.json | 28 ++++---- src/daq_queuing_service/api/api.py | 26 ++++--- .../blueapi/blueapi_call.py | 6 +- .../external_interaction/tiled/tiled.py | 4 +- src/daq_queuing_service/plugins/converter.py | 47 ++++++++---- .../plugins/i15_1/i15_1_converter.py | 19 ++--- src/daq_queuing_service/task_queue/queue.py | 69 +++++++++--------- src/daq_queuing_service/task_queue/task.py | 6 +- tests/unit_tests/conftest.py | 10 +-- .../plugins/i15_1/test_i15_1_converter.py | 18 ++--- tests/unit_tests/plugins/test_converter.py | 8 ++- tests/unit_tests/test_api.py | 22 +++--- tests/unit_tests/test_queue.py | 72 +++++++++---------- 14 files changed, 187 insertions(+), 164 deletions(-) diff --git a/docs/explanations/converters.md b/docs/explanations/converters.md index 3a8c48f4..06ed4f5d 100644 --- a/docs/explanations/converters.md +++ b/docs/explanations/converters.md @@ -15,25 +15,23 @@ Subclass `Converter` and override the methods you need: ```python from daq_queuing_service.plugins.converter import Converter + class MyConverter(Converter): - def validate(self, experiments: list[TaskRequest | Experiment]): - ... + def validate(self, experiments: list[TaskRequest | Experiment]): ... def pre_process( self, queue: list[Task], - history: list[TaskWithPosition], + history: list[QueueTask], call_history: list[BlueapiCall], - ) -> list[Task]: - ... + ) -> list[Task]: ... def construct_blueapi_calls( self, - queue: list[TaskWithPosition], - history: list[TaskWithPosition], + queue: list[QueueTask], + history: list[QueueTask], call_history: list[BlueapiCall], - ) -> list[BlueapiCall]: - ... + ) -> list[BlueapiCall]: ... ``` ### `validate()` diff --git a/docs/reference/rest_api.json b/docs/reference/rest_api.json index 1f2d33ce..522db877 100644 --- a/docs/reference/rest_api.json +++ b/docs/reference/rest_api.json @@ -139,7 +139,7 @@ "schema": { "type": "array", "items": { - "$ref": "#/components/schemas/TaskWithPosition" + "$ref": "#/components/schemas/QueueTask" }, "title": "Response Get Queued Tasks Queue Get" } @@ -238,7 +238,7 @@ "schema": { "type": "array", "items": { - "$ref": "#/components/schemas/TaskWithPosition" + "$ref": "#/components/schemas/QueueTask" }, "title": "Response Cancel All Tasks Queue Delete" } @@ -318,7 +318,7 @@ "application/json": { "schema": { "items": { - "$ref": "#/components/schemas/TaskWithPosition" + "$ref": "#/components/schemas/QueueTask" }, "type": "array", "title": "Response Cancel Tasks Queue Tasks Delete" @@ -362,7 +362,7 @@ "schema": { "anyOf": [ { - "$ref": "#/components/schemas/TaskWithPosition" + "$ref": "#/components/schemas/QueueTask" }, { "type": "null" @@ -416,7 +416,7 @@ "schema": { "type": "array", "items": { - "$ref": "#/components/schemas/TaskWithPosition" + "$ref": "#/components/schemas/QueueTask" }, "title": "Response Get All Tasks Tasks Get" } @@ -457,7 +457,7 @@ "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/TaskWithPosition" + "$ref": "#/components/schemas/QueueTask" } } } @@ -505,7 +505,7 @@ "schema": { "type": "array", "items": { - "$ref": "#/components/schemas/TaskWithPosition" + "$ref": "#/components/schemas/QueueTask" }, "title": "Response Get Completed Tasks History Get" } @@ -550,7 +550,7 @@ "application/json": { "schema": { "items": { - "$ref": "#/components/schemas/BlueapiCallResponse" + "$ref": "#/components/schemas/QueueBlueapiCall" }, "type": "array", "title": "Response Get Call Queue Call Queue Get" @@ -572,7 +572,7 @@ "application/json": { "schema": { "items": { - "$ref": "#/components/schemas/BlueapiCallResponse" + "$ref": "#/components/schemas/QueueBlueapiCall" }, "type": "array", "title": "Response Get Call History Call History Get" @@ -678,7 +678,7 @@ "title": "BasicAuthentication", "description": "User credentials for basic authentication" }, - "BlueapiCallResponse": { + "QueueBlueapiCall": { "properties": { "task_request": { "$ref": "#/components/schemas/TaskRequest" @@ -782,7 +782,7 @@ "tiled_ids", "scan_ids" ], - "title": "BlueapiCallResponse" + "title": "QueueBlueapiCall" }, "BlueapiConfig": { "properties": { @@ -1320,7 +1320,7 @@ "title": "TaskResult", "description": "Serializable wrapper around the result of a plan\n\nIf the result is not serializable, the result will be None but the type\nwill be the name of the type. If the result is actually None, the type will\nbe 'NoneType'." }, - "TaskWithPosition": { + "QueueTask": { "properties": { "experiment": { "anyOf": [ @@ -1342,7 +1342,7 @@ }, "blueapi_calls": { "items": { - "$ref": "#/components/schemas/BlueapiCallResponse" + "$ref": "#/components/schemas/QueueBlueapiCall" }, "type": "array", "title": "Blueapi Calls" @@ -1382,7 +1382,7 @@ "kind", "user" ], - "title": "TaskWithPosition" + "title": "QueueTask" }, "User": { "properties": { diff --git a/src/daq_queuing_service/api/api.py b/src/daq_queuing_service/api/api.py index 1cbb849e..263b4774 100644 --- a/src/daq_queuing_service/api/api.py +++ b/src/daq_queuing_service/api/api.py @@ -13,7 +13,7 @@ from daq_queuing_service.app.config import AppConfig from daq_queuing_service.broadcaster import Broadcaster from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( - BlueapiCallResponse, + QueueBlueapiCall, ) from daq_queuing_service.log import LOGGER from daq_queuing_service.plugins.converter import Converter, ValidateError @@ -21,8 +21,8 @@ QUEUE_EVENTS, PauseReason, QueueState, + QueueTask, TaskQueue, - TaskWithPosition, ) from daq_queuing_service.task_queue.task import Experiment, Status, Task @@ -45,9 +45,7 @@ class TaskCancelRequest(BaseModel): task_ids: list[str] -def _filter_by_status( - tasks: list[TaskWithPosition], status: Status | None -) -> list[TaskWithPosition]: +def _filter_by_status(tasks: list[QueueTask], status: Status | None) -> list[QueueTask]: if status is None: return tasks return [task for task in tasks if task.status == status] @@ -99,7 +97,7 @@ async def update_queue_state(payload: QueueStateUpdate) -> QueueState: return await queue.resume_queue() @router.get("/queue") - async def get_queued_tasks(status: Status | None = None) -> list[TaskWithPosition]: + async def get_queued_tasks(status: Status | None = None) -> list[QueueTask]: return _filter_by_status(await queue.get_queue(), status) @router.post("/queue") @@ -119,7 +117,7 @@ async def add_tasks_to_queue( return task_ids @router.delete("/queue") - async def cancel_all_tasks() -> list[TaskWithPosition]: + async def cancel_all_tasks() -> list[QueueTask]: return await queue.cancel_all_tasks() @router.post("/queue/move") @@ -127,25 +125,25 @@ async def move_task(task_id: str, new_position: int) -> int: return await queue.move_task(task_id, new_position) @router.delete("/queue/tasks") - async def cancel_tasks(payload: TaskCancelRequest) -> list[TaskWithPosition]: + async def cancel_tasks(payload: TaskCancelRequest) -> list[QueueTask]: return await queue.cancel_tasks(payload.task_ids) @router.get("/queue/{position}") - async def get_task_by_position(position: int) -> TaskWithPosition | None: + async def get_task_by_position(position: int) -> QueueTask | None: return await queue.get_task_by_position(position) @router.get("/tasks") - async def get_all_tasks(status: Status | None = None) -> list[TaskWithPosition]: + async def get_all_tasks(status: Status | None = None) -> list[QueueTask]: return _filter_by_status(await queue.get_tasks(), status) @router.get("/tasks/{task_id}") - async def get_task_by_id(task_id: str) -> TaskWithPosition: + async def get_task_by_id(task_id: str) -> QueueTask: return await queue.get_task_by_id(task_id) @router.get("/history") async def get_completed_tasks( status: Status | None = None, - ) -> list[TaskWithPosition]: + ) -> list[QueueTask]: return _filter_by_status(await queue.get_history(), status) @router.delete("/history") @@ -153,11 +151,11 @@ async def clear_history(): return await queue.clear_history() @router.get("/call_queue") - async def get_call_queue() -> list[BlueapiCallResponse]: + async def get_call_queue() -> list[QueueBlueapiCall]: return await queue.get_call_queue() @router.get("/call_history") - async def get_call_history() -> list[BlueapiCallResponse]: + async def get_call_history() -> list[QueueBlueapiCall]: return await queue.get_call_history() @router.get("/events") diff --git a/src/daq_queuing_service/external_interaction/blueapi/blueapi_call.py b/src/daq_queuing_service/external_interaction/blueapi/blueapi_call.py index e7f34b66..82f78400 100644 --- a/src/daq_queuing_service/external_interaction/blueapi/blueapi_call.py +++ b/src/daq_queuing_service/external_interaction/blueapi/blueapi_call.py @@ -103,11 +103,11 @@ def fail(self, errors: list[str | TaskError] | None = None): def skip(self): self._update_status(CallStatus.SKIPPED) - def to_response(self) -> "BlueapiCallResponse": - return BlueapiCallResponse.from_blueapi_call(self) + def to_response(self) -> "QueueBlueapiCall": + return QueueBlueapiCall.from_blueapi_call(self) -class BlueapiCallResponse(BaseModel): +class QueueBlueapiCall(BaseModel): task_request: TaskRequest parent_task_id: str | None status: CallStatus diff --git a/src/daq_queuing_service/external_interaction/tiled/tiled.py b/src/daq_queuing_service/external_interaction/tiled/tiled.py index 3e3b7b96..b695dff1 100644 --- a/src/daq_queuing_service/external_interaction/tiled/tiled.py +++ b/src/daq_queuing_service/external_interaction/tiled/tiled.py @@ -9,7 +9,9 @@ from tiled.client.container import Container from tiled.queries import Eq -from daq_queuing_service.external_interaction.blueapi.blueapi_call import BlueapiCall +from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( + BlueapiCall, +) from daq_queuing_service.log import LOGGER # Ignoring the following rule as the tiled client is poorly typed diff --git a/src/daq_queuing_service/plugins/converter.py b/src/daq_queuing_service/plugins/converter.py index 8e889330..27dc8c43 100644 --- a/src/daq_queuing_service/plugins/converter.py +++ b/src/daq_queuing_service/plugins/converter.py @@ -5,11 +5,12 @@ from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( BlueapiCall, + QueueBlueapiCall, ) from daq_queuing_service.task_queue.task import ( Experiment, + QueueTask, Task, - TaskWithPosition, ) @@ -41,9 +42,9 @@ def validate(self, experiments: list[TaskRequest | Experiment]) -> None: def pre_process( self, - current_task: TaskWithPosition | None, + current_task: QueueTask | None, queue: list[Task], - history: list[TaskWithPosition], + history: list[QueueTask], call_history: list[BlueapiCall], ) -> list[Task]: """This gets run whenever something changes in the queue, before @@ -57,10 +58,10 @@ def pre_process( If not overridden, this method returns the queue as is with no modification. Args: - current_task (TaskWithPosition | None): The current running task, if one + current_task (QueueTask | None): The current running task, if one exists. queue (list[Task]): List of tasks currently in the queue - history (list[TaskWithPosition]): List of completed tasks + history (list[QueueTask]): List of completed tasks call_history (list[BlueapiCall]): List of completed blueapi calls Returns: @@ -70,8 +71,8 @@ def pre_process( def construct_blueapi_calls( self, - queue: list[TaskWithPosition], - history: list[TaskWithPosition], + queue: list[QueueTask], + history: list[QueueTask], call_history: list[BlueapiCall], ) -> list[BlueapiCall]: """Converts the list of queued tasks into a list of blueapi calls. This is @@ -81,8 +82,8 @@ def construct_blueapi_calls( implementation does. Args: - queue (list[TaskWithPosition]): List of tasks in the queue - history (list[TaskWithPosition]): List of completed tasks + queue (list[QueueTask]): List of tasks in the queue + history (list[QueueTask]): List of completed tasks call_history (list[BlueapiCall]): List of completed blueapi calls Returns: @@ -99,12 +100,12 @@ def construct_blueapi_calls( return call_list def pause_on_complete_blueapi_call( - self, call: BlueapiCall, result: TaskResult + self, call: QueueBlueapiCall, result: TaskResult ) -> bool: """Decide whether to pause the queue after a blueapi call completes successfully Args: - call (BlueapiCall): The completed call + call (QueueBlueapiCall): The completed call result (TaskResult): The result of the call from blueapi Returns: @@ -113,12 +114,12 @@ def pause_on_complete_blueapi_call( return False def pause_on_fail_blueapi_call( - self, call: BlueapiCall, errors: list[str | TaskError] | None + self, call: QueueBlueapiCall, errors: list[str | TaskError] | None ) -> bool: """Decide whether to pause the queue after a blueapi call fails Args: - call (BlueapiCall): The failed call + call (QueueBlueapiCall): The failed call errors (list[str | TaskError] | None): The errors that occurred Returns: @@ -126,6 +127,26 @@ def pause_on_fail_blueapi_call( """ return True + def process_running_task( + self, + completed_plans: list[QueueBlueapiCall], + plans_to_do: list[BlueapiCall], + ) -> list[BlueapiCall]: + """An opportunity to process the currently running task, by amending the + blueapi calls yet to be made. Get's called right after a blueapi call completes. + + Args: + completed_plans (list[QueueBlueapiCall]): Completed blueapi calls + associated with the currently running task. + plans_to_do (list[BlueapiCall]): Future blueapi calls associated with the + currently running task. + + Returns: + list[BlueapiCall]: An amended list of future blueapi calls for the running + task. + """ + return plans_to_do + def _construct_blueapi_task_request( self, experiment: Experiment | TaskRequest, diff --git a/src/daq_queuing_service/plugins/i15_1/i15_1_converter.py b/src/daq_queuing_service/plugins/i15_1/i15_1_converter.py index d323cb45..c9e5cee2 100644 --- a/src/daq_queuing_service/plugins/i15_1/i15_1_converter.py +++ b/src/daq_queuing_service/plugins/i15_1/i15_1_converter.py @@ -8,6 +8,7 @@ from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( BlueapiCall, + QueueBlueapiCall, ) from daq_queuing_service.external_interaction.tiled.tiled import get_tiled_client from daq_queuing_service.log import ( @@ -35,9 +36,9 @@ ContainerPosition, Experiment, ExperimentDefinition, + QueueTask, Sample, Task, - TaskWithPosition, ) CONFIG_SERVER_URL = "https://i15-1-daq-config.diamond.ac.uk" @@ -84,9 +85,9 @@ def _queue_params(self) -> QueueParameters: def pre_process( self, - current_task: TaskWithPosition | None, + current_task: QueueTask | None, queue: list[Task], - history: list[TaskWithPosition], + history: list[QueueTask], call_history: list[BlueapiCall], ) -> list[Task]: self._config_client.reset_cache() @@ -96,8 +97,8 @@ def pre_process( def construct_blueapi_calls( self, - queue: list[TaskWithPosition], - history: list[TaskWithPosition], + queue: list[QueueTask], + history: list[QueueTask], call_history: list[BlueapiCall], ) -> list[BlueapiCall]: call_list: list[BlueapiCall] = [] @@ -127,7 +128,7 @@ def construct_blueapi_calls( return call_list def pause_on_fail_blueapi_call( - self, call: BlueapiCall, errors: list[str | TaskError] | None + self, call: QueueBlueapiCall, errors: list[str | TaskError] | None ) -> bool: for error in errors or []: @@ -177,13 +178,13 @@ def _construct_blueapi_tasks_from_experiment( return plans def _add_required_auxiliary_scans( - self, current_task: TaskWithPosition | None, tasks: list[Task] + self, current_task: QueueTask | None, tasks: list[Task] ) -> list[Task]: """Adds auxiliary scan tasks to the queue. They will be added directly in front of the first task in the queue that requires them. Args: - current_task (TaskWithPosition | None): Current running task, if one exists. + current_task (QueueTask | None): Current running task, if one exists. tasks (list[Task]): Current list of tasks Returns: @@ -221,7 +222,7 @@ def _add_required_auxiliary_scans( def _ensure_auxiliary_in_queue_or_tiled( self, required_auxiliary: AuxiliaryScan, - current_task: TaskWithPosition | None, + current_task: QueueTask | None, new_tasks: list[Task], task_id: str, instrument_session: str, diff --git a/src/daq_queuing_service/task_queue/queue.py b/src/daq_queuing_service/task_queue/queue.py index bba131ec..4809e12f 100644 --- a/src/daq_queuing_service/task_queue/queue.py +++ b/src/daq_queuing_service/task_queue/queue.py @@ -13,8 +13,8 @@ from daq_queuing_service.broadcaster import Broadcaster, Event from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( BlueapiCall, - BlueapiCallResponse, CallStatus, + QueueBlueapiCall, ) from daq_queuing_service.external_interaction.tiled.tiled import ( get_tiled_and_scan_ids, @@ -35,7 +35,7 @@ TaskNotFoundError, TaskNotInQueueError, ) -from daq_queuing_service.task_queue.task import Status, Task, TaskWithPosition +from daq_queuing_service.task_queue.task import QueueTask, Status, Task class TaskRegistry(dict[str, Task]): @@ -58,8 +58,8 @@ class QueueState(BaseModel): class QueueContents(TypedDict): """A class to represent the contents of the queue - tasks (dict[str, Task]): A dict containing all the tasks contained in the queue, - including history, with the task ID as the key. + tasks (dict[str, Task]): A dict containing all the tasks contained in the + queue, including history, with the task ID as the key. queue (list[str]): A list of the task IDs of the tasks in the queue. The order of the list represents the order of the queue. @@ -338,7 +338,7 @@ async def complete_call(self, call: BlueapiCall, result: TaskResult): with add_session_to_logger(call.task_request.instrument_session): tiled_ids, scan_ids = get_tiled_and_scan_ids(self._tiled_client, call) should_pause = self._converter.pause_on_complete_blueapi_call( - call.model_copy(), result + call.to_response(), result ) LOGGER.info(f"Plan {call} completed successfully. Result: {result}") @@ -363,7 +363,7 @@ async def fail_call( """ with add_session_to_logger(call.task_request.instrument_session): should_pause = self._converter.pause_on_fail_blueapi_call( - call.model_copy(), errors + call.to_response(), errors ) tiled_ids, scan_ids = get_tiled_and_scan_ids(self._tiled_client, call) LOGGER.error(f"Plan {call} has failed with the following errors: {errors}") @@ -382,14 +382,14 @@ async def fail_call( call.fail(errors) self._call_history.append(call) - async def get_task_by_id(self, task_id: str) -> TaskWithPosition: + async def get_task_by_id(self, task_id: str) -> QueueTask: """Returns a task based on it's task ID Args: task_id (str): Task ID of the task Returns: - TaskWithPosition: A copy of the task with a position field. + QueueTask: A copy of the task with a position field. Raises: TaskNotFoundError: Raised if the no task exists with the requested task ID. @@ -398,57 +398,57 @@ async def get_task_by_id(self, task_id: str) -> TaskWithPosition: async with self._lock: return self._get_task_by_id(task_id) - def _get_task_by_id(self, task_id: str) -> TaskWithPosition: + def _get_task_by_id(self, task_id: str) -> QueueTask: task = self._tasks[task_id] position = self._queue.index(task.id) if task.id in self._queue else None - return TaskWithPosition.from_task(task, position) + return QueueTask.from_task(task, position) - async def get_task_by_position(self, position: int) -> TaskWithPosition | None: + async def get_task_by_position(self, position: int) -> QueueTask | None: """Return a task based on it's position in the queue Args: position (int): The position of the task to be returned. Returns: - TaskWithPosition | None: A copy of the task with a position field, or None + QueueTask | None: A copy of the task with a position field, or None if no task exists at the requested position. """ # Returns copy so don't have to be worried about caller modifying task. async with self._lock: return self._get_task_by_position(position) - def _get_task_by_position(self, position: int) -> TaskWithPosition | None: + def _get_task_by_position(self, position: int) -> QueueTask | None: if position < -self.length or position >= self.length: return return self._get_task_by_id(self._queue[position]) - async def get_queue(self) -> list[TaskWithPosition]: + async def get_queue(self) -> list[QueueTask]: """Get the entire queue (not including history) Returns: - list[TaskWithPosition]: A list of the tasks in the queue, in the order they + list[QueueTask]: A list of the tasks in the queue, in the order they will be run in. """ # Returns copies so don't have to be worried about caller modifying tasks. async with self._lock: return self._get_queue() - async def get_history(self) -> list[TaskWithPosition]: + async def get_history(self) -> list[QueueTask]: """Get the history list. Returns: - list[TaskWithPosition]: A list of the tasks in the history list, in + list[QueueTask]: A list of the tasks in the history list, in chronological order. """ # Returns copies so don't have to be worried about caller modifying tasks. async with self._lock: return self._get_history() - async def get_tasks(self) -> list[TaskWithPosition]: + async def get_tasks(self) -> list[QueueTask]: """Get all the tasks in the queue and history list. Returns: - list[TaskWithPosition]: A list of tasks in chronological order, starting + list[QueueTask]: A list of tasks in chronological order, starting with the history. """ # Returns copies so don't have to be worried about caller modifying tasks. @@ -494,7 +494,7 @@ async def move_task(self, task_id: str, position: int) -> int: LOGGER.info(f"Successfully moved task {task_id} to position {new_position}") return new_position - async def cancel_tasks(self, task_ids: Sequence[str]) -> list[TaskWithPosition]: + async def cancel_tasks(self, task_ids: Sequence[str]) -> list[QueueTask]: """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. @@ -503,7 +503,7 @@ async def cancel_tasks(self, task_ids: Sequence[str]) -> list[TaskWithPosition]: task_ids (Sequence[str]): List of task IDs to cancel Returns: - list[TaskWithPosition]: List of the task objects that were removed from the + list[QueueTask]: List of the task objects that were removed from the queue. """ async with self._modifying: @@ -511,7 +511,7 @@ async def cancel_tasks(self, task_ids: Sequence[str]) -> list[TaskWithPosition]: LOGGER.info(f"Successfully cancelled tasks: {task_ids}") return cancelled_tasks - async def cancel_all_tasks(self) -> list[TaskWithPosition]: + async def cancel_all_tasks(self) -> list[QueueTask]: async with self._modifying: task_ids = [ id @@ -624,14 +624,14 @@ def _add_tasks(self, tasks: list[Task], position: int | None) -> None: for task in tasks: self._tasks[task.id] = task - def _cancel_tasks(self, task_ids: Sequence[str]) -> list[TaskWithPosition]: + def _cancel_tasks(self, task_ids: Sequence[str]) -> list[QueueTask]: task_ids = list(task_ids) self._validate_tasks_for_move_or_deletion(task_ids) self._remove_tasks_from_queue(task_ids) tasks = [self._tasks[task_id] for task_id in task_ids] for task in tasks: task.cancel() - return [TaskWithPosition.from_task(task) for task in tasks] + return [QueueTask.from_task(task) for task in tasks] def _remove_tasks_from_queue(self, task_ids: list[str]) -> list[str]: # Only removes tasks in the queue (not history or registry) @@ -676,33 +676,30 @@ def _validate_tasks_for_move_or_deletion(self, task_ids: list[str]): f"Cannot move task '{task_id}', it is currently in progress!" ) - def _get_queue(self) -> list[TaskWithPosition]: + def _get_queue(self) -> list[QueueTask]: return [ - TaskWithPosition.from_task(self._tasks[task_id], i) + QueueTask.from_task(self._tasks[task_id], i) for i, task_id in enumerate(self._queue) ] - def _get_history(self) -> list[TaskWithPosition]: - return [ - TaskWithPosition.from_task(self._tasks[task_id]) - for task_id in self._history - ] + def _get_history(self) -> list[QueueTask]: + return [QueueTask.from_task(self._tasks[task_id]) for task_id in self._history] - async def get_call_queue(self) -> list[BlueapiCallResponse]: + async def get_call_queue(self) -> list[QueueBlueapiCall]: async with self._lock: return self._get_call_queue() - def _get_call_queue(self) -> list[BlueapiCallResponse]: + def _get_call_queue(self) -> list[QueueBlueapiCall]: return [call.to_response() for call in self._call_queue] - async def get_call_history(self) -> list[BlueapiCallResponse]: + async def get_call_history(self) -> list[QueueBlueapiCall]: async with self._lock: return self._get_call_history() - def _get_call_history(self) -> list[BlueapiCallResponse]: + def _get_call_history(self) -> list[QueueBlueapiCall]: return [call.to_response() for call in self._call_history] - def _get_running_task(self) -> TaskWithPosition | None: + def _get_running_task(self) -> QueueTask | None: current_task = self._get_task_by_position(0) if current_task and current_task.status == Status.IN_PROGRESS: return current_task diff --git a/src/daq_queuing_service/task_queue/task.py b/src/daq_queuing_service/task_queue/task.py index ce5bb477..2f5da8da 100644 --- a/src/daq_queuing_service/task_queue/task.py +++ b/src/daq_queuing_service/task_queue/task.py @@ -8,8 +8,8 @@ from daq_queuing_service.app.authentication import User from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( BlueapiCall, - BlueapiCallResponse, CallStatus, + QueueBlueapiCall, ) @@ -103,11 +103,11 @@ def kind(self) -> TaskKind: return TaskKind.PLAN -class TaskWithPosition(BaseModel): +class QueueTask(BaseModel): experiment: Experiment | TaskRequest id: str status: Status - blueapi_calls: list[BlueapiCallResponse] + blueapi_calls: list[QueueBlueapiCall] position: int | None kind: TaskKind user: User | None diff --git a/tests/unit_tests/conftest.py b/tests/unit_tests/conftest.py index fa59f1fd..cb7dbd43 100644 --- a/tests/unit_tests/conftest.py +++ b/tests/unit_tests/conftest.py @@ -19,7 +19,9 @@ load_config, ) from daq_queuing_service.broadcaster import Broadcaster -from daq_queuing_service.external_interaction.blueapi.blueapi_call import BlueapiCall +from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( + BlueapiCall, +) from daq_queuing_service.external_interaction.tiled.tiled import cache from daq_queuing_service.log import LOGGER, setup_logger from daq_queuing_service.plugins.converter import Converter @@ -29,9 +31,9 @@ ContainerPosition, Experiment, ExperimentDefinition, + QueueTask, Sample, Task, - TaskWithPosition, ) from daq_queuing_service.worker.worker import LOGGER as WORKER_LOGGER @@ -106,8 +108,8 @@ async def task_queue(tasks: list[Task], converter: Converter): @pytest.fixture async def task_queue_one_to_many(tasks: list[Task]): def construct_blueapi_calls( - queue: list[TaskWithPosition], - history: list[TaskWithPosition], + queue: list[QueueTask], + history: list[QueueTask], call_history: list[BlueapiCall], ) -> list[BlueapiCall]: call_list: list[BlueapiCall] = [] diff --git a/tests/unit_tests/plugins/i15_1/test_i15_1_converter.py b/tests/unit_tests/plugins/i15_1/test_i15_1_converter.py index f0685504..7feac5b0 100644 --- a/tests/unit_tests/plugins/i15_1/test_i15_1_converter.py +++ b/tests/unit_tests/plugins/i15_1/test_i15_1_converter.py @@ -27,10 +27,10 @@ from daq_queuing_service.task_queue.task import ( Experiment, ExperimentDefinition, + QueueTask, Status, Task, TaskKind, - TaskWithPosition, ) from ...conftest import make_sample @@ -75,7 +75,7 @@ def make_background_task(capillary: STANDARD_CAPILLARY, time_per_pdf: int) -> Ta ) -def assert_tasks_equal(task1: Task | TaskWithPosition, task2: Task | TaskWithPosition): +def assert_tasks_equal(task1: Task | QueueTask, task2: Task | QueueTask): # Check two tasks are equal other than the generated UUID copy1 = type(task1).model_validate(task1) copy2 = type(task2).model_validate(task2) @@ -345,7 +345,7 @@ def test_experiment_with_correct_experiment_type_are_converted(): sample=make_sample("test_8_1", ""), instrument_session="cm12345-1", ) - task = TaskWithPosition( + task = QueueTask( experiment=experiment, id="1", status=Status.QUEUED, @@ -517,7 +517,7 @@ def test_mix_of_experiments_with_correct_experiment_type_are_converted(): sample=make_sample("test_8_1", ""), instrument_session="cm12345-1", ) - good_task = TaskWithPosition( + good_task = QueueTask( experiment=good_experiment, id="1", status=Status.QUEUED, @@ -765,7 +765,7 @@ def test__ensure_background_in_queue_or_tiled_returns_if_current_task_is_suitabl new_tasks: list[Task] = [] result = i15_1_converter._ensure_auxiliary_in_queue_or_tiled( background, - TaskWithPosition.from_task(make_background_task("fq1.0", 25)), + QueueTask.from_task(make_background_task("fq1.0", 25)), new_tasks, "task_id", "cm12345-1", @@ -781,7 +781,7 @@ def test__ensure_background_in_queue_or_tiled_modifies_queued_background_if_poss pin=StandardsPin(capillary="fq1.0", contents=None), time_per_pdf=25, ) - current_task = TaskWithPosition.from_task(make_background_task("fq1.0", 25)) + current_task = QueueTask.from_task(make_background_task("fq1.0", 25)) # Current task is suitable but not a background assert isinstance(current_task.experiment, Experiment) current_task.experiment.experiment_definition.name = "Not a background" @@ -926,7 +926,7 @@ def test_no_robot_load_queued_if_turned_off_in_queue_params( queue_params.do_robot_load = False blueapi_calls = I151Converter().construct_blueapi_calls( - [TaskWithPosition.from_task(i15_1_tasks[0])], [], [] + [QueueTask.from_task(i15_1_tasks[0])], [], [] ) assert not any(call.task_request.name == "robot_load" for call in blueapi_calls) assert not any(call.task_request.name == "robot_unload" for call in blueapi_calls) @@ -939,7 +939,7 @@ def test_no_centring_queued_if_turned_off_in_queue_params( queue_params.do_centring = False blueapi_calls = I151Converter().construct_blueapi_calls( - [TaskWithPosition.from_task(i15_1_tasks[0])], [], [] + [QueueTask.from_task(i15_1_tasks[0])], [], [] ) assert not any(call.task_request.name == "centre_sample" for call in blueapi_calls) wait_for_beams = [ @@ -955,6 +955,6 @@ def test_no_wait_for_beam_queued_if_turned_off_in_queue_params( queue_params.do_wait_for_beam = False blueapi_calls = I151Converter().construct_blueapi_calls( - [TaskWithPosition.from_task(i15_1_tasks[0])], [], [] + [QueueTask.from_task(i15_1_tasks[0])], [], [] ) assert not any(call.task_request.name == "wait_for_beam" for call in blueapi_calls) diff --git a/tests/unit_tests/plugins/test_converter.py b/tests/unit_tests/plugins/test_converter.py index 24715322..f99566e7 100644 --- a/tests/unit_tests/plugins/test_converter.py +++ b/tests/unit_tests/plugins/test_converter.py @@ -1,7 +1,9 @@ import pytest from blueapi.service.model import TaskRequest -from daq_queuing_service.external_interaction.blueapi.blueapi_call import BlueapiCall +from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( + BlueapiCall, +) from daq_queuing_service.plugins.converter import ( Converter, get_converter, @@ -9,8 +11,8 @@ from daq_queuing_service.task_queue.task import ( Experiment, ExperimentDefinition, + QueueTask, Task, - TaskWithPosition, ) from ..conftest import make_sample @@ -50,7 +52,7 @@ def test_default_converter_pre_process_does_not_modify_queue(tasks: list[Task]): def test_default_converter_construct_blueapi_calls_creates_one_blueapi_call_per_task( bluesky_tasks: list[Task], ): - task_copies = [TaskWithPosition.from_task(task) for task in bluesky_tasks] + task_copies = [QueueTask.from_task(task) for task in bluesky_tasks] blueapi_calls = Converter().construct_blueapi_calls(task_copies, [], []) for task, blueapi_call in zip(task_copies, blueapi_calls, strict=True): assert isinstance(task.experiment, TaskRequest) diff --git a/tests/unit_tests/test_api.py b/tests/unit_tests/test_api.py index 5250521a..7a85d7b1 100644 --- a/tests/unit_tests/test_api.py +++ b/tests/unit_tests/test_api.py @@ -31,18 +31,18 @@ from daq_queuing_service.broadcaster import Broadcaster, Event from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( BlueapiCall, - BlueapiCallResponse, CallStatus, + QueueBlueapiCall, ) from daq_queuing_service.plugins.converter import Converter from daq_queuing_service.task_queue.queue import QUEUE_EVENTS, PauseReason, TaskQueue from daq_queuing_service.task_queue.queue_utils import QueueError from daq_queuing_service.task_queue.task import ( Experiment, + QueueTask, Status, Task, TaskKind, - TaskWithPosition, ) @@ -393,7 +393,7 @@ async def test_add_tasks_to_queue_adds_to_queue_and_and_returns_task_ids( task_ids: list[str] = response.json() [uuid.UUID(task_id) for task_id in task_ids] - assert await task_queue_with_history.get_task_by_position(-1) == TaskWithPosition( + assert await task_queue_with_history.get_task_by_position(-1) == QueueTask( experiment=TaskRequest( name="add_tasks", params={"time": 10}, @@ -401,7 +401,7 @@ async def test_add_tasks_to_queue_adds_to_queue_and_and_returns_task_ids( ), id=task_ids[-1], blueapi_calls=[ - BlueapiCallResponse( + QueueBlueapiCall( task_request=TaskRequest( name="add_tasks", params={"time": 10}, instrument_session="abc" ), @@ -475,9 +475,9 @@ async def test_if_sync_fails_after_tasks_added_then_contents_restored_and_error( class SomeError(Exception): ... def fail_conversion( - current_task: TaskWithPosition | None, + current_task: QueueTask | None, queue: list[Task], - history: list[TaskWithPosition], + history: list[QueueTask], call_history: list[BlueapiCall], ): raise SomeError("Conversion failed because xyz") @@ -610,7 +610,10 @@ async def test_move_task_moves_task_and_returns_position( "0", 5, 409, - {"error": "task_not_in_queue", "message": "Task 0 isn't present in queue"}, + { + "error": "task_not_in_queue", + "message": "Task 0 isn't present in queue", + }, ], [ "2", @@ -779,7 +782,10 @@ async def test_cancel_tasks_removes_task_from_queue_and_returns_tasks( [ TaskCancelRequest(task_ids=["1", "2", "3", "4"]).model_dump(), 409, - {"error": "task_not_in_queue", "message": "Task 1 isn't present in queue"}, + { + "error": "task_not_in_queue", + "message": "Task 1 isn't present in queue", + }, ], ], ) diff --git a/tests/unit_tests/test_queue.py b/tests/unit_tests/test_queue.py index 7a9cd792..cc67464f 100644 --- a/tests/unit_tests/test_queue.py +++ b/tests/unit_tests/test_queue.py @@ -10,17 +10,17 @@ from daq_queuing_service.broadcaster import Broadcaster from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( BlueapiCall, - BlueapiCallResponse, CallStatus, + QueueBlueapiCall, ) from daq_queuing_service.plugins.converter import Converter, ConverterError from daq_queuing_service.task_queue.queue import ( PauseReason, QueueContents, QueueState, + QueueTask, TaskQueue, TaskRegistry, - TaskWithPosition, ) from daq_queuing_service.task_queue.queue_utils import ( NegativePositionError, @@ -281,7 +281,7 @@ async def test_get_queue_only_returns_tasks_in_queue( assert task_queue_with_history._history == ["0", "1"] result = await task_queue_with_history.get_queue() assert result == [ - TaskWithPosition( + QueueTask( experiment=Experiment( name="test_experiment", instrument_session="", @@ -293,7 +293,7 @@ async def test_get_queue_only_returns_tasks_in_queue( id="2", status=Status.IN_PROGRESS, blueapi_calls=[ - BlueapiCallResponse( + QueueBlueapiCall( task_request=TaskRequest(name="test", instrument_session=""), parent_task_id="2", status=CallStatus.IN_PROGRESS, @@ -310,7 +310,7 @@ async def test_get_queue_only_returns_tasks_in_queue( kind=TaskKind.EXPERIMENT, user=None, ), - TaskWithPosition( + QueueTask( experiment=Experiment( name="test_experiment", instrument_session="", @@ -322,7 +322,7 @@ async def test_get_queue_only_returns_tasks_in_queue( id="3", status=Status.QUEUED, blueapi_calls=[ - BlueapiCallResponse( + QueueBlueapiCall( task_request=TaskRequest(name="test", instrument_session=""), parent_task_id="3", status=CallStatus.WAITING, @@ -339,7 +339,7 @@ async def test_get_queue_only_returns_tasks_in_queue( kind=TaskKind.EXPERIMENT, user=None, ), - TaskWithPosition( + QueueTask( experiment=Experiment( name="test_experiment", instrument_session="", @@ -351,7 +351,7 @@ async def test_get_queue_only_returns_tasks_in_queue( id="4", status=Status.QUEUED, blueapi_calls=[ - BlueapiCallResponse( + QueueBlueapiCall( task_request=TaskRequest(name="test", instrument_session=""), parent_task_id="4", status=CallStatus.WAITING, @@ -377,7 +377,7 @@ async def test_get_history_only_returns_tasks_in_history( assert task_queue_with_history._queue == ["2", "3", "4"] result = await task_queue_with_history.get_history() assert result == [ - TaskWithPosition( + QueueTask( experiment=Experiment( name="test_experiment", instrument_session="", @@ -389,7 +389,7 @@ async def test_get_history_only_returns_tasks_in_history( id="0", status=Status.ERROR, blueapi_calls=[ - BlueapiCallResponse( + QueueBlueapiCall( task_request=TaskRequest(name="test", instrument_session=""), parent_task_id="0", status=CallStatus.ERROR, @@ -412,7 +412,7 @@ async def test_get_history_only_returns_tasks_in_history( kind=TaskKind.EXPERIMENT, user=None, ), - TaskWithPosition( + QueueTask( experiment=Experiment( name="test_experiment", instrument_session="", @@ -424,7 +424,7 @@ async def test_get_history_only_returns_tasks_in_history( id="1", status=Status.COMPLETE, blueapi_calls=[ - BlueapiCallResponse( + QueueBlueapiCall( task_request=TaskRequest(name="test", instrument_session=""), parent_task_id="1", status=CallStatus.SUCCESS, @@ -451,7 +451,7 @@ async def test_get_tasks_returns_tasks_in_queue_and_history( assert task_queue_with_history._history == ["0", "1"] result = await task_queue_with_history.get_tasks() assert result == [ - TaskWithPosition( + QueueTask( experiment=Experiment( name="test_experiment", instrument_session="", @@ -463,7 +463,7 @@ async def test_get_tasks_returns_tasks_in_queue_and_history( id="0", status=Status.ERROR, blueapi_calls=[ - BlueapiCallResponse( + QueueBlueapiCall( task_request=TaskRequest(name="test", instrument_session=""), parent_task_id="0", status=CallStatus.ERROR, @@ -486,7 +486,7 @@ async def test_get_tasks_returns_tasks_in_queue_and_history( kind=TaskKind.EXPERIMENT, user=None, ), - TaskWithPosition( + QueueTask( experiment=Experiment( name="test_experiment", instrument_session="", @@ -498,7 +498,7 @@ async def test_get_tasks_returns_tasks_in_queue_and_history( id="1", status=Status.COMPLETE, blueapi_calls=[ - BlueapiCallResponse( + QueueBlueapiCall( task_request=TaskRequest(name="test", instrument_session=""), parent_task_id="1", status=CallStatus.SUCCESS, @@ -515,7 +515,7 @@ async def test_get_tasks_returns_tasks_in_queue_and_history( kind=TaskKind.EXPERIMENT, user=None, ), - TaskWithPosition( + QueueTask( experiment=Experiment( name="test_experiment", instrument_session="", @@ -527,7 +527,7 @@ async def test_get_tasks_returns_tasks_in_queue_and_history( id="2", status=Status.IN_PROGRESS, blueapi_calls=[ - BlueapiCallResponse( + QueueBlueapiCall( task_request=TaskRequest(name="test", instrument_session=""), parent_task_id="2", status=CallStatus.IN_PROGRESS, @@ -544,7 +544,7 @@ async def test_get_tasks_returns_tasks_in_queue_and_history( kind=TaskKind.EXPERIMENT, user=None, ), - TaskWithPosition( + QueueTask( experiment=Experiment( name="test_experiment", instrument_session="", @@ -556,7 +556,7 @@ async def test_get_tasks_returns_tasks_in_queue_and_history( id="3", status=Status.QUEUED, blueapi_calls=[ - BlueapiCallResponse( + QueueBlueapiCall( task_request=TaskRequest(name="test", instrument_session=""), parent_task_id="3", status=CallStatus.WAITING, @@ -573,7 +573,7 @@ async def test_get_tasks_returns_tasks_in_queue_and_history( kind=TaskKind.EXPERIMENT, user=None, ), - TaskWithPosition( + QueueTask( experiment=Experiment( name="test_experiment", instrument_session="", @@ -585,7 +585,7 @@ async def test_get_tasks_returns_tasks_in_queue_and_history( id="4", status=Status.QUEUED, blueapi_calls=[ - BlueapiCallResponse( + QueueBlueapiCall( task_request=TaskRequest(name="test", instrument_session=""), parent_task_id="4", status=CallStatus.WAITING, @@ -616,12 +616,8 @@ async def test_get_task_by_id_returns_task_in_queue_or_history( "4" in task_queue_with_history._queue and "4" not in task_queue_with_history._history ) - assert isinstance( - await task_queue_with_history.get_task_by_id("0"), TaskWithPosition - ) - assert isinstance( - await task_queue_with_history.get_task_by_id("4"), TaskWithPosition - ) + assert isinstance(await task_queue_with_history.get_task_by_id("0"), QueueTask) + assert isinstance(await task_queue_with_history.get_task_by_id("4"), QueueTask) async def test_get_task_by_id_raises_error_if_task_id_does_not_exist( @@ -635,7 +631,7 @@ async def test_get_task_by_pos_returns_task_in_queue( task_queue_with_history: TaskQueue, ): task = await task_queue_with_history.get_task_by_position(2) - assert isinstance(task, TaskWithPosition) + assert isinstance(task, QueueTask) assert task.position == 2 @@ -934,7 +930,7 @@ async def test_get_call_queue_returns_calls_in_call_queue( ): result = await task_queue_with_history.get_call_queue() assert result == [ - BlueapiCallResponse( + QueueBlueapiCall( task_request=TaskRequest(name="test", params={}, instrument_session=""), parent_task_id="2", status=CallStatus.IN_PROGRESS, @@ -946,7 +942,7 @@ async def test_get_call_queue_returns_calls_in_call_queue( tiled_ids=[], scan_ids=[], ), - BlueapiCallResponse( + QueueBlueapiCall( task_request=TaskRequest(name="test", params={}, instrument_session=""), parent_task_id="3", status=CallStatus.WAITING, @@ -958,7 +954,7 @@ async def test_get_call_queue_returns_calls_in_call_queue( tiled_ids=[], scan_ids=[], ), - BlueapiCallResponse( + QueueBlueapiCall( task_request=TaskRequest(name="test", params={}, instrument_session=""), parent_task_id="4", status=CallStatus.WAITING, @@ -978,7 +974,7 @@ async def test_get_call_history_returns_calls_in_call_history( ): result = await task_queue_with_history.get_call_history() assert result == [ - BlueapiCallResponse( + QueueBlueapiCall( task_request=TaskRequest(name="test", params={}, instrument_session=""), parent_task_id="0", status=CallStatus.ERROR, @@ -994,7 +990,7 @@ async def test_get_call_history_returns_calls_in_call_history( tiled_ids=[], scan_ids=[], ), - BlueapiCallResponse( + QueueBlueapiCall( task_request=TaskRequest(name="test", params={}, instrument_session=""), parent_task_id="1", status=CallStatus.SUCCESS, @@ -1212,8 +1208,8 @@ async def test_if_error_during_conversion_then_error_handled_and_contents_restor task_queue: TaskQueue, ): def convert( - queue: list[TaskWithPosition], - history: list[TaskWithPosition], + queue: list[QueueTask], + history: list[QueueTask], call_history: list[BlueapiCall], ): for task_id in task_queue._queue: @@ -1235,8 +1231,8 @@ async def test_if_error_during_conversion_then__restore_latest_good_contents_cal task_queue: TaskQueue, ): def convert( - queue: list[TaskWithPosition], - history: list[TaskWithPosition], + queue: list[QueueTask], + history: list[QueueTask], call_history: list[BlueapiCall], ): raise ValueError("Conversion failed") From 05c304d19b6dd96a765b5f139979334f6f614dc5 Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Mon, 2 Nov 2026 09:15:42 +0000 Subject: [PATCH 02/11] Update api docs --- docs/reference/rest_api.json | 365 ++++++++++++++++++----------------- 1 file changed, 192 insertions(+), 173 deletions(-) diff --git a/docs/reference/rest_api.json b/docs/reference/rest_api.json index 522db877..cda4da03 100644 --- a/docs/reference/rest_api.json +++ b/docs/reference/rest_api.json @@ -678,112 +678,6 @@ "title": "BasicAuthentication", "description": "User credentials for basic authentication" }, - "QueueBlueapiCall": { - "properties": { - "task_request": { - "$ref": "#/components/schemas/TaskRequest" - }, - "parent_task_id": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "title": "Parent Task Id" - }, - "status": { - "$ref": "#/components/schemas/CallStatus" - }, - "time_started": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "title": "Time Started" - }, - "time_completed": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "title": "Time Completed" - }, - "result": { - "anyOf": [ - { - "$ref": "#/components/schemas/TaskResult" - }, - { - "type": "null" - } - ] - }, - "errors": { - "items": { - "anyOf": [ - { - "type": "string" - }, - { - "$ref": "#/components/schemas/TaskError" - } - ] - }, - "type": "array", - "title": "Errors" - }, - "blueapi_id": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "title": "Blueapi Id" - }, - "tiled_ids": { - "items": { - "type": "string" - }, - "type": "array", - "title": "Tiled Ids" - }, - "scan_ids": { - "items": { - "type": "integer" - }, - "type": "array", - "title": "Scan Ids" - } - }, - "type": "object", - "required": [ - "task_request", - "parent_task_id", - "status", - "time_started", - "time_completed", - "result", - "errors", - "blueapi_id", - "tiled_ids", - "scan_ids" - ], - "title": "QueueBlueapiCall" - }, "BlueapiConfig": { "properties": { "stomp": { @@ -1048,9 +942,29 @@ "OIDCConfig": { "properties": { "well_known_url": { - "type": "string", + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], "title": "Well Known Url", - "description": "URL to fetch OIDC config from the provider" + "description": "URL to fetch OIDC config from the provider", + "deprecated": true + }, + "issuer": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Issuer", + "description": "URL of OIDC provider" }, "client_id": { "type": "string", @@ -1073,7 +987,6 @@ "additionalProperties": false, "type": "object", "required": [ - "well_known_url", "client_id" ], "title": "OIDCConfig" @@ -1088,6 +1001,112 @@ ], "title": "PauseReason" }, + "QueueBlueapiCall": { + "properties": { + "task_request": { + "$ref": "#/components/schemas/TaskRequest" + }, + "parent_task_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Parent Task Id" + }, + "status": { + "$ref": "#/components/schemas/CallStatus" + }, + "time_started": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Time Started" + }, + "time_completed": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Time Completed" + }, + "result": { + "anyOf": [ + { + "$ref": "#/components/schemas/TaskResult" + }, + { + "type": "null" + } + ] + }, + "errors": { + "items": { + "anyOf": [ + { + "type": "string" + }, + { + "$ref": "#/components/schemas/TaskError" + } + ] + }, + "type": "array", + "title": "Errors" + }, + "blueapi_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Blueapi Id" + }, + "tiled_ids": { + "items": { + "type": "string" + }, + "type": "array", + "title": "Tiled Ids" + }, + "scan_ids": { + "items": { + "type": "integer" + }, + "type": "array", + "title": "Scan Ids" + } + }, + "type": "object", + "required": [ + "task_request", + "parent_task_id", + "status", + "time_started", + "time_completed", + "result", + "errors", + "blueapi_id", + "tiled_ids", + "scan_ids" + ], + "title": "QueueBlueapiCall" + }, "QueueState": { "properties": { "paused": { @@ -1118,6 +1137,70 @@ ], "title": "QueueStateUpdate" }, + "QueueTask": { + "properties": { + "experiment": { + "anyOf": [ + { + "$ref": "#/components/schemas/Experiment" + }, + { + "$ref": "#/components/schemas/TaskRequest" + } + ], + "title": "Experiment" + }, + "id": { + "type": "string", + "title": "Id" + }, + "status": { + "$ref": "#/components/schemas/Status" + }, + "blueapi_calls": { + "items": { + "$ref": "#/components/schemas/QueueBlueapiCall" + }, + "type": "array", + "title": "Blueapi Calls" + }, + "position": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Position" + }, + "kind": { + "$ref": "#/components/schemas/TaskKind" + }, + "user": { + "anyOf": [ + { + "$ref": "#/components/schemas/User" + }, + { + "type": "null" + } + ] + } + }, + "type": "object", + "required": [ + "experiment", + "id", + "status", + "blueapi_calls", + "position", + "kind", + "user" + ], + "title": "QueueTask" + }, "RestConfig": { "properties": { "url": { @@ -1320,70 +1403,6 @@ "title": "TaskResult", "description": "Serializable wrapper around the result of a plan\n\nIf the result is not serializable, the result will be None but the type\nwill be the name of the type. If the result is actually None, the type will\nbe 'NoneType'." }, - "QueueTask": { - "properties": { - "experiment": { - "anyOf": [ - { - "$ref": "#/components/schemas/Experiment" - }, - { - "$ref": "#/components/schemas/TaskRequest" - } - ], - "title": "Experiment" - }, - "id": { - "type": "string", - "title": "Id" - }, - "status": { - "$ref": "#/components/schemas/Status" - }, - "blueapi_calls": { - "items": { - "$ref": "#/components/schemas/QueueBlueapiCall" - }, - "type": "array", - "title": "Blueapi Calls" - }, - "position": { - "anyOf": [ - { - "type": "integer" - }, - { - "type": "null" - } - ], - "title": "Position" - }, - "kind": { - "$ref": "#/components/schemas/TaskKind" - }, - "user": { - "anyOf": [ - { - "$ref": "#/components/schemas/User" - }, - { - "type": "null" - } - ] - } - }, - "type": "object", - "required": [ - "experiment", - "id", - "status", - "blueapi_calls", - "position", - "kind", - "user" - ], - "title": "QueueTask" - }, "User": { "properties": { "fedid": { From 88b434ec43905d1b8ba0ede8d17bc66b5be5d281 Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Tue, 3 Nov 2026 16:42:42 +0000 Subject: [PATCH 03/11] Refactor to add inheritance --- docs/reference/rest_api.json | 63 ++++++---------- .../blueapi/blueapi_call.py | 38 +++------- src/daq_queuing_service/task_queue/queue.py | 8 +- src/daq_queuing_service/task_queue/task.py | 75 +++++++++---------- .../plugins/i15_1/test_i15_1_converter.py | 10 +-- tests/unit_tests/test_queue.py | 36 --------- 6 files changed, 77 insertions(+), 153 deletions(-) diff --git a/docs/reference/rest_api.json b/docs/reference/rest_api.json index 5caa5da5..a7cde4e8 100644 --- a/docs/reference/rest_api.json +++ b/docs/reference/rest_api.json @@ -998,18 +998,12 @@ "$ref": "#/components/schemas/TaskRequest" }, "parent_task_id": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], + "type": "string", "title": "Parent Task Id" }, "status": { - "$ref": "#/components/schemas/CallStatus" + "$ref": "#/components/schemas/CallStatus", + "default": "Waiting" }, "time_started": { "anyOf": [ @@ -1093,21 +1087,14 @@ "type": "null" } ], - "title": "Runtime" + "title": "Runtime", + "readOnly": true } }, "type": "object", "required": [ "task_request", "parent_task_id", - "status", - "time_started", - "time_completed", - "result", - "errors", - "blueapi_id", - "tiled_ids", - "scan_ids", "runtime" ], "title": "QueueBlueapiCall" @@ -1170,9 +1157,6 @@ "type": "string", "title": "Id" }, - "status": { - "$ref": "#/components/schemas/Status" - }, "blueapi_calls": { "items": { "$ref": "#/components/schemas/QueueBlueapiCall" @@ -1180,31 +1164,28 @@ "type": "array", "title": "Blueapi Calls" }, - "position": { + "user": { "anyOf": [ { - "type": "integer" + "$ref": "#/components/schemas/User" }, { "type": "null" } - ], - "title": "Position" - }, - "kind": { - "$ref": "#/components/schemas/TaskKind" + ] }, - "user": { + "estimated_runtime": { "anyOf": [ { - "$ref": "#/components/schemas/User" + "type": "integer" }, { "type": "null" } - ] + ], + "title": "Estimated Runtime" }, - "estimated_runtime": { + "position": { "anyOf": [ { "type": "integer" @@ -1213,7 +1194,14 @@ "type": "null" } ], - "title": "Estimated Runtime" + "title": "Position" + }, + "status": { + "$ref": "#/components/schemas/Status" + }, + "kind": { + "$ref": "#/components/schemas/TaskKind", + "readOnly": true }, "runtime": { "anyOf": [ @@ -1224,19 +1212,16 @@ "type": "null" } ], - "title": "Runtime" + "title": "Runtime", + "readOnly": true } }, "type": "object", "required": [ "experiment", - "id", - "status", - "blueapi_calls", "position", + "status", "kind", - "user", - "estimated_runtime", "runtime" ], "title": "QueueTask" diff --git a/src/daq_queuing_service/external_interaction/blueapi/blueapi_call.py b/src/daq_queuing_service/external_interaction/blueapi/blueapi_call.py index e251a54e..a664a09a 100644 --- a/src/daq_queuing_service/external_interaction/blueapi/blueapi_call.py +++ b/src/daq_queuing_service/external_interaction/blueapi/blueapi_call.py @@ -31,7 +31,7 @@ def allowed_transitions(self): return allowed_transitions[self] -class BlueapiCall(BaseModel): +class QueueBlueapiCall(BaseModel): task_request: TaskRequest parent_task_id: str status: CallStatus = CallStatus.WAITING @@ -43,6 +43,14 @@ class BlueapiCall(BaseModel): tiled_ids: list[str] = Field(default_factory=list) scan_ids: list[int] = Field(default_factory=list[int]) + @computed_field + @property + def runtime(self) -> float | None: + if self.time_completed is not None and self.time_started is not None: + return self.time_completed.timestamp() - self.time_started.timestamp() + + +class BlueapiCall(QueueBlueapiCall): def _update_status(self, new_status: CallStatus): """Updates the status of the task, checking that the transition is valid @@ -103,29 +111,5 @@ def fail(self, errors: list[str | TaskError] | None = None): def skip(self): self._update_status(CallStatus.SKIPPED) - def to_response(self) -> "QueueBlueapiCall": - return QueueBlueapiCall.from_blueapi_call(self) - - @computed_field - @property - def runtime(self) -> float | None: - if self.time_completed is not None and self.time_started is not None: - return self.time_completed.timestamp() - self.time_started.timestamp() - - -class QueueBlueapiCall(BaseModel): - task_request: TaskRequest - parent_task_id: str | None - status: CallStatus - time_started: datetime | None - time_completed: datetime | None - result: TaskResult | None - errors: list[str | TaskError] - blueapi_id: str | None - tiled_ids: list[str] - scan_ids: list[int] - runtime: float | None - - @classmethod - def from_blueapi_call(cls, blueapi_call: BlueapiCall): - return cls.model_validate({**blueapi_call.model_dump()}) + def to_parent(self) -> "QueueBlueapiCall": + return QueueBlueapiCall.model_validate(self.model_dump()) diff --git a/src/daq_queuing_service/task_queue/queue.py b/src/daq_queuing_service/task_queue/queue.py index c6c654a4..6225984c 100644 --- a/src/daq_queuing_service/task_queue/queue.py +++ b/src/daq_queuing_service/task_queue/queue.py @@ -356,7 +356,7 @@ async def complete_call(self, call: BlueapiCall, result: TaskResult): with add_session_to_logger(call.task_request.instrument_session): tiled_ids, scan_ids = get_tiled_and_scan_ids(self._tiled_client, call) should_pause = self._converter.pause_on_complete_blueapi_call( - call.to_response(), result + call.to_parent(), result ) LOGGER.info(f"Plan {call} completed successfully. Result: {result}") @@ -381,7 +381,7 @@ async def fail_call( """ with add_session_to_logger(call.task_request.instrument_session): should_pause = self._converter.pause_on_fail_blueapi_call( - call.to_response(), errors + call.to_parent(), errors ) tiled_ids, scan_ids = get_tiled_and_scan_ids(self._tiled_client, call) LOGGER.error(f"Plan {call} has failed with the following errors: {errors}") @@ -710,14 +710,14 @@ async def get_call_queue(self) -> list[QueueBlueapiCall]: return self._get_call_queue() def _get_call_queue(self) -> list[QueueBlueapiCall]: - return [call.to_response() for call in self._call_queue] + return [call.to_parent() for call in self._call_queue] async def get_call_history(self) -> list[QueueBlueapiCall]: async with self._lock: return self._get_call_history() def _get_call_history(self) -> list[QueueBlueapiCall]: - return [call.to_response() for call in self._call_history] + return [call.to_parent() for call in self._call_history] def _get_running_task(self) -> QueueTask | None: current_task = self._get_task_by_position(0) diff --git a/src/daq_queuing_service/task_queue/task.py b/src/daq_queuing_service/task_queue/task.py index e7a01e44..e49c8087 100644 --- a/src/daq_queuing_service/task_queue/task.py +++ b/src/daq_queuing_service/task_queue/task.py @@ -1,9 +1,9 @@ from enum import StrEnum -from typing import Any, Self +from typing import Any, Generic, Self, TypeVar from uuid import uuid4 from blueapi.service.model import TaskRequest -from pydantic import BaseModel, Field, computed_field +from pydantic import BaseModel, Field, PrivateAttr, computed_field from daq_queuing_service.app.authentication import User from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( @@ -60,13 +60,41 @@ class TaskKind(StrEnum): PLAN = "Plan" -class Task(BaseModel): +T = TypeVar("T", bound=QueueBlueapiCall) + + +class BaseTask(BaseModel, Generic[T]): experiment: Experiment | TaskRequest - id: str = Field(default_factory=create_uuid_str) - blueapi_calls: list[BlueapiCall] = Field(default_factory=lambda: []) - user: User | None = None + id: str = Field(default_factory=create_uuid_str, frozen=True) + blueapi_calls: list[T] = Field(default_factory=lambda: []) + user: User | None = Field(default=None, frozen=True) estimated_runtime: int | None = None - _cancelled: bool = False + + @computed_field + @property + def kind(self) -> TaskKind: + match self.experiment: + case Experiment(): + return TaskKind.EXPERIMENT + case TaskRequest(): + return TaskKind.PLAN + + @computed_field + @property + def runtime(self) -> float | None: + # This should get nicer once we have separate types for queued, in progress and + # complete tasks. + if runtimes := [blueapi_call.runtime for blueapi_call in self.blueapi_calls]: + total = 0 + for runtime in runtimes: + if runtime is None: + return + total += runtime + return total + + +class Task(BaseTask[BlueapiCall]): + _cancelled: bool = PrivateAttr(False) def cancel(self): self._cancelled = True @@ -94,39 +122,10 @@ def status(self) -> Status: return Status.IN_PROGRESS return Status.QUEUED - @computed_field - @property - def kind(self) -> TaskKind: - match self.experiment: - case Experiment(): - return TaskKind.EXPERIMENT - case TaskRequest(): - return TaskKind.PLAN - - @computed_field - @property - def runtime(self) -> float | None: - # This should get nicer once we have separate types for queued, in progress and - # complete tasks. - if runtimes := [blueapi_call.runtime for blueapi_call in self.blueapi_calls]: - total = 0 - for runtime in runtimes: - if runtime is None: - return - total += runtime - return total - -class QueueTask(BaseModel): - experiment: Experiment | TaskRequest - id: str - status: Status - blueapi_calls: list[QueueBlueapiCall] +class QueueTask(BaseTask[QueueBlueapiCall]): position: int | None - kind: TaskKind - user: User | None - estimated_runtime: int | None - runtime: float | None + status: Status @classmethod def from_task(cls, task: Task, position: int | None = None) -> Self: diff --git a/tests/unit_tests/plugins/i15_1/test_i15_1_converter.py b/tests/unit_tests/plugins/i15_1/test_i15_1_converter.py index f59ff909..ed30823a 100644 --- a/tests/unit_tests/plugins/i15_1/test_i15_1_converter.py +++ b/tests/unit_tests/plugins/i15_1/test_i15_1_converter.py @@ -77,10 +77,7 @@ def make_background_task(capillary: STANDARD_CAPILLARY, time_per_pdf: int) -> Ta def assert_tasks_equal(task1: Task | QueueTask, task2: Task | QueueTask): # Check two tasks are equal other than the generated UUID - copy1 = type(task1).model_validate(task1) - copy2 = type(task2).model_validate(task2) - copy1.id = copy2.id = "" - assert task1 == task2 + assert task1.model_dump(exclude={"id"}) == task2.model_dump(exclude={"id"}) def create_task_dict(task_id: str, experiment: dict[str, Any]) -> dict[str, Any]: @@ -353,9 +350,7 @@ def test_experiment_with_correct_experiment_type_are_converted(): status=Status.QUEUED, blueapi_calls=[], position=None, - kind=TaskKind.EXPERIMENT, user=None, - runtime=None, estimated_runtime=100, ) call_list = I151Converter().construct_blueapi_calls([task], [], []) @@ -527,9 +522,7 @@ def test_mix_of_experiments_with_correct_experiment_type_are_converted(): status=Status.QUEUED, blueapi_calls=[], position=None, - kind=TaskKind.EXPERIMENT, user=None, - runtime=None, estimated_runtime=100, ) @@ -540,7 +533,6 @@ class BadExperiment: bad_task.experiment = BadExperiment() # type: ignore plan_task = deepcopy(good_task) plan_task.experiment = TaskRequest(name="", instrument_session="") - plan_task.kind = TaskKind.PLAN tasks = [good_task, bad_task, plan_task, good_task] call_list = I151Converter().construct_blueapi_calls(tasks, [], []) assert len(call_list) == 13 diff --git a/tests/unit_tests/test_queue.py b/tests/unit_tests/test_queue.py index 726a2ccd..c7a49e57 100644 --- a/tests/unit_tests/test_queue.py +++ b/tests/unit_tests/test_queue.py @@ -37,7 +37,6 @@ ExperimentDefinition, Status, Task, - TaskKind, ) from .conftest import make_sample @@ -307,13 +306,10 @@ async def test_get_queue_only_returns_tasks_in_queue( blueapi_id=None, tiled_ids=[], scan_ids=[], - runtime=None, ) ], position=0, - kind=TaskKind.EXPERIMENT, user=None, - runtime=None, estimated_runtime=100, ), QueueTask( @@ -339,13 +335,10 @@ async def test_get_queue_only_returns_tasks_in_queue( blueapi_id=None, tiled_ids=[], scan_ids=[], - runtime=None, ) ], position=1, - kind=TaskKind.EXPERIMENT, user=None, - runtime=None, estimated_runtime=100, ), QueueTask( @@ -371,13 +364,10 @@ async def test_get_queue_only_returns_tasks_in_queue( blueapi_id=None, tiled_ids=[], scan_ids=[], - runtime=None, ) ], position=2, - kind=TaskKind.EXPERIMENT, user=None, - runtime=None, estimated_runtime=100, ), ] @@ -418,13 +408,10 @@ async def test_get_history_only_returns_tasks_in_history( blueapi_id=None, tiled_ids=[], scan_ids=[], - runtime=59, ) ], position=None, - kind=TaskKind.EXPERIMENT, user=None, - runtime=59, estimated_runtime=100, ), QueueTask( @@ -450,13 +437,10 @@ async def test_get_history_only_returns_tasks_in_history( blueapi_id=None, tiled_ids=[], scan_ids=[], - runtime=59, ) ], position=None, - kind=TaskKind.EXPERIMENT, user=None, - runtime=59, estimated_runtime=100, ), ] @@ -498,13 +482,10 @@ async def test_get_tasks_returns_tasks_in_queue_and_history( blueapi_id=None, tiled_ids=[], scan_ids=[], - runtime=59, ) ], position=None, - kind=TaskKind.EXPERIMENT, user=None, - runtime=59, estimated_runtime=100, ), QueueTask( @@ -530,13 +511,10 @@ async def test_get_tasks_returns_tasks_in_queue_and_history( blueapi_id=None, tiled_ids=[], scan_ids=[], - runtime=59, ) ], position=None, - kind=TaskKind.EXPERIMENT, user=None, - runtime=59, estimated_runtime=100, ), QueueTask( @@ -562,13 +540,10 @@ async def test_get_tasks_returns_tasks_in_queue_and_history( blueapi_id=None, tiled_ids=[], scan_ids=[], - runtime=None, ) ], position=0, - kind=TaskKind.EXPERIMENT, user=None, - runtime=None, estimated_runtime=100, ), QueueTask( @@ -594,13 +569,10 @@ async def test_get_tasks_returns_tasks_in_queue_and_history( blueapi_id=None, tiled_ids=[], scan_ids=[], - runtime=None, ) ], position=1, - kind=TaskKind.EXPERIMENT, user=None, - runtime=None, estimated_runtime=100, ), QueueTask( @@ -626,13 +598,10 @@ async def test_get_tasks_returns_tasks_in_queue_and_history( blueapi_id=None, tiled_ids=[], scan_ids=[], - runtime=None, ) ], position=2, - kind=TaskKind.EXPERIMENT, user=None, - runtime=None, estimated_runtime=100, ), ] @@ -976,7 +945,6 @@ async def test_get_call_queue_returns_calls_in_call_queue( blueapi_id=None, tiled_ids=[], scan_ids=[], - runtime=None, ), QueueBlueapiCall( task_request=TaskRequest(name="test", params={}, instrument_session=""), @@ -989,7 +957,6 @@ async def test_get_call_queue_returns_calls_in_call_queue( blueapi_id=None, tiled_ids=[], scan_ids=[], - runtime=None, ), QueueBlueapiCall( task_request=TaskRequest(name="test", params={}, instrument_session=""), @@ -1002,7 +969,6 @@ async def test_get_call_queue_returns_calls_in_call_queue( blueapi_id=None, tiled_ids=[], scan_ids=[], - runtime=None, ), ] @@ -1027,7 +993,6 @@ async def test_get_call_history_returns_calls_in_call_history( blueapi_id=None, tiled_ids=[], scan_ids=[], - runtime=59, ), QueueBlueapiCall( task_request=TaskRequest(name="test", params={}, instrument_session=""), @@ -1040,7 +1005,6 @@ async def test_get_call_history_returns_calls_in_call_history( blueapi_id=None, tiled_ids=[], scan_ids=[], - runtime=59, ), ] From df2b4df5bdd91a5c9127f4f91d1316e7ba621637 Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Tue, 3 Nov 2026 17:41:53 +0000 Subject: [PATCH 04/11] Tidy and remove changes from other PR --- .../blueapi/blueapi_call.py | 4 +++ src/daq_queuing_service/plugins/converter.py | 29 +++---------------- .../plugins/i15_1/i15_1_converter.py | 13 ++++----- src/daq_queuing_service/task_queue/queue.py | 29 ++++++++++--------- tests/unit_tests/plugins/test_converter.py | 4 +-- tests/unit_tests/test_queue.py | 9 ++++-- 6 files changed, 38 insertions(+), 50 deletions(-) diff --git a/src/daq_queuing_service/external_interaction/blueapi/blueapi_call.py b/src/daq_queuing_service/external_interaction/blueapi/blueapi_call.py index a664a09a..c1638204 100644 --- a/src/daq_queuing_service/external_interaction/blueapi/blueapi_call.py +++ b/src/daq_queuing_service/external_interaction/blueapi/blueapi_call.py @@ -113,3 +113,7 @@ def skip(self): def to_parent(self) -> "QueueBlueapiCall": return QueueBlueapiCall.model_validate(self.model_dump()) + + @classmethod + def from_parent(cls, queue_blueapi_call: QueueBlueapiCall) -> "BlueapiCall": + return cls.model_validate(queue_blueapi_call.model_dump()) diff --git a/src/daq_queuing_service/plugins/converter.py b/src/daq_queuing_service/plugins/converter.py index 27dc8c43..eccc510e 100644 --- a/src/daq_queuing_service/plugins/converter.py +++ b/src/daq_queuing_service/plugins/converter.py @@ -5,7 +5,6 @@ from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( BlueapiCall, - QueueBlueapiCall, ) from daq_queuing_service.task_queue.task import ( Experiment, @@ -100,12 +99,12 @@ def construct_blueapi_calls( return call_list def pause_on_complete_blueapi_call( - self, call: QueueBlueapiCall, result: TaskResult + self, call: BlueapiCall, result: TaskResult ) -> bool: """Decide whether to pause the queue after a blueapi call completes successfully Args: - call (QueueBlueapiCall): The completed call + call (BlueapiCall): The completed call result (TaskResult): The result of the call from blueapi Returns: @@ -114,12 +113,12 @@ def pause_on_complete_blueapi_call( return False def pause_on_fail_blueapi_call( - self, call: QueueBlueapiCall, errors: list[str | TaskError] | None + self, call: BlueapiCall, errors: list[str | TaskError] | None ) -> bool: """Decide whether to pause the queue after a blueapi call fails Args: - call (QueueBlueapiCall): The failed call + call (BlueapiCall): The failed call errors (list[str | TaskError] | None): The errors that occurred Returns: @@ -127,26 +126,6 @@ def pause_on_fail_blueapi_call( """ return True - def process_running_task( - self, - completed_plans: list[QueueBlueapiCall], - plans_to_do: list[BlueapiCall], - ) -> list[BlueapiCall]: - """An opportunity to process the currently running task, by amending the - blueapi calls yet to be made. Get's called right after a blueapi call completes. - - Args: - completed_plans (list[QueueBlueapiCall]): Completed blueapi calls - associated with the currently running task. - plans_to_do (list[BlueapiCall]): Future blueapi calls associated with the - currently running task. - - Returns: - list[BlueapiCall]: An amended list of future blueapi calls for the running - task. - """ - return plans_to_do - def _construct_blueapi_task_request( self, experiment: Experiment | TaskRequest, diff --git a/src/daq_queuing_service/plugins/i15_1/i15_1_converter.py b/src/daq_queuing_service/plugins/i15_1/i15_1_converter.py index c9e5cee2..2a13a0f7 100644 --- a/src/daq_queuing_service/plugins/i15_1/i15_1_converter.py +++ b/src/daq_queuing_service/plugins/i15_1/i15_1_converter.py @@ -7,7 +7,6 @@ from tiled.client.container import Container as TiledContainer from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( - BlueapiCall, QueueBlueapiCall, ) from daq_queuing_service.external_interaction.tiled.tiled import get_tiled_client @@ -88,7 +87,7 @@ def pre_process( current_task: QueueTask | None, queue: list[Task], history: list[QueueTask], - call_history: list[BlueapiCall], + call_history: list[QueueBlueapiCall], ) -> list[Task]: self._config_client.reset_cache() if self._queue_params.add_auxiliary_scans: @@ -99,23 +98,23 @@ def construct_blueapi_calls( self, queue: list[QueueTask], history: list[QueueTask], - call_history: list[BlueapiCall], - ) -> list[BlueapiCall]: - call_list: list[BlueapiCall] = [] + call_history: list[QueueBlueapiCall], + ) -> list[QueueBlueapiCall]: + call_list: list[QueueBlueapiCall] = [] for task in queue: with add_session_to_logger(task.experiment.instrument_session): match task.experiment: case TaskRequest(): call_list.append( - BlueapiCall( + QueueBlueapiCall( task_request=task.experiment, parent_task_id=task.id ) ) case Experiment(): call_list.extend( [ - BlueapiCall( + QueueBlueapiCall( task_request=b_api_task, parent_task_id=task.id ) for b_api_task in ( diff --git a/src/daq_queuing_service/task_queue/queue.py b/src/daq_queuing_service/task_queue/queue.py index 6225984c..6de9aa84 100644 --- a/src/daq_queuing_service/task_queue/queue.py +++ b/src/daq_queuing_service/task_queue/queue.py @@ -222,21 +222,24 @@ def _sync(self): ] try: - new_calls = self._converter.construct_blueapi_calls( - [ - task - for task in self._get_queue_with_positions() - if task.status == Status.QUEUED - ], - self._get_history(), - self._queue_history, - ) + new_blueapi_calls = [ + BlueapiCall.from_parent(call) + for call in self._converter.construct_blueapi_calls( + [ + task + for task in self._get_queue_with_positions() + if task.status == Status.QUEUED + ], + self._get_history(), + self._queue_history, + ) + ] except Exception as e: raise ConverterError(e) from e - self._call_queue.extend(new_calls) + self._call_queue.extend(new_blueapi_calls) - for call in new_calls: + for call in new_blueapi_calls: # Add children to parent tasks if call.parent_task_id: self._tasks[call.parent_task_id].blueapi_calls.append(call) @@ -356,7 +359,7 @@ async def complete_call(self, call: BlueapiCall, result: TaskResult): with add_session_to_logger(call.task_request.instrument_session): tiled_ids, scan_ids = get_tiled_and_scan_ids(self._tiled_client, call) should_pause = self._converter.pause_on_complete_blueapi_call( - call.to_parent(), result + call.model_copy(), result ) LOGGER.info(f"Plan {call} completed successfully. Result: {result}") @@ -381,7 +384,7 @@ async def fail_call( """ with add_session_to_logger(call.task_request.instrument_session): should_pause = self._converter.pause_on_fail_blueapi_call( - call.to_parent(), errors + call.model_copy(), errors ) tiled_ids, scan_ids = get_tiled_and_scan_ids(self._tiled_client, call) LOGGER.error(f"Plan {call} has failed with the following errors: {errors}") diff --git a/tests/unit_tests/plugins/test_converter.py b/tests/unit_tests/plugins/test_converter.py index f99566e7..29ff67ef 100644 --- a/tests/unit_tests/plugins/test_converter.py +++ b/tests/unit_tests/plugins/test_converter.py @@ -2,7 +2,7 @@ from blueapi.service.model import TaskRequest from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( - BlueapiCall, + QueueBlueapiCall, ) from daq_queuing_service.plugins.converter import ( Converter, @@ -56,6 +56,6 @@ def test_default_converter_construct_blueapi_calls_creates_one_blueapi_call_per_ blueapi_calls = Converter().construct_blueapi_calls(task_copies, [], []) for task, blueapi_call in zip(task_copies, blueapi_calls, strict=True): assert isinstance(task.experiment, TaskRequest) - assert blueapi_call == BlueapiCall( + assert blueapi_call == QueueBlueapiCall( task_request=task.experiment, parent_task_id=task.id ) diff --git a/tests/unit_tests/test_queue.py b/tests/unit_tests/test_queue.py index c7a49e57..af9c2229 100644 --- a/tests/unit_tests/test_queue.py +++ b/tests/unit_tests/test_queue.py @@ -1214,7 +1214,7 @@ async def test_if_error_during_conversion_then_error_handled_and_contents_restor def convert( queue: list[QueueTask], history: list[QueueTask], - call_history: list[BlueapiCall], + call_history: list[QueueBlueapiCall], ): for task_id in task_queue._queue: del task_queue._tasks[task_id] @@ -1237,7 +1237,7 @@ async def test_if_error_during_conversion_then__restore_latest_good_contents_cal def convert( queue: list[QueueTask], history: list[QueueTask], - call_history: list[BlueapiCall], + call_history: list[QueueBlueapiCall], ): raise ValueError("Conversion failed") @@ -1310,10 +1310,13 @@ async def test__sync_calls_converter_pre_process_with_expected_args( for task_id in task_queue_with_history._queue[1:] ] history = task_queue_with_history._get_history() + call_history = task_queue_with_history._get_call_history() task_queue_with_history._sync() - converter.pre_process.assert_called_once_with(first_task, other_tasks, history, []) + converter.pre_process.assert_called_once_with( + first_task, other_tasks, history, call_history + ) async def test_complete_call_gets_md_from_tiled_and_adds_to_call_object( From 27318bce085b10cfa50e72d36ae56f273f93fca5 Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Tue, 3 Nov 2026 17:50:11 +0000 Subject: [PATCH 05/11] Fix bug --- src/daq_queuing_service/plugins/converter.py | 9 +++++---- .../plugins/i15_1/i15_1_converter.py | 11 ++++++----- src/daq_queuing_service/task_queue/queue.py | 5 ++--- tests/unit_tests/test_queue.py | 4 ++-- 4 files changed, 15 insertions(+), 14 deletions(-) diff --git a/src/daq_queuing_service/plugins/converter.py b/src/daq_queuing_service/plugins/converter.py index eccc510e..6496f602 100644 --- a/src/daq_queuing_service/plugins/converter.py +++ b/src/daq_queuing_service/plugins/converter.py @@ -5,6 +5,7 @@ from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( BlueapiCall, + QueueBlueapiCall, ) from daq_queuing_service.task_queue.task import ( Experiment, @@ -44,7 +45,7 @@ def pre_process( current_task: QueueTask | None, queue: list[Task], history: list[QueueTask], - call_history: list[BlueapiCall], + call_history: list[QueueBlueapiCall], ) -> list[Task]: """This gets run whenever something changes in the queue, before 'construct_blueapi_calls'. It is an opportunity to automatically modify the @@ -61,7 +62,7 @@ def pre_process( exists. queue (list[Task]): List of tasks currently in the queue history (list[QueueTask]): List of completed tasks - call_history (list[BlueapiCall]): List of completed blueapi calls + call_history (list[QueueBlueapiCall]): List of completed blueapi calls Returns: list[Task]: The new list of tasks in the queue @@ -72,7 +73,7 @@ def construct_blueapi_calls( self, queue: list[QueueTask], history: list[QueueTask], - call_history: list[BlueapiCall], + call_history: list[QueueBlueapiCall], ) -> list[BlueapiCall]: """Converts the list of queued tasks into a list of blueapi calls. This is needed if the queue contains ulims experiments, as these must be mapped onto @@ -83,7 +84,7 @@ def construct_blueapi_calls( Args: queue (list[QueueTask]): List of tasks in the queue history (list[QueueTask]): List of completed tasks - call_history (list[BlueapiCall]): List of completed blueapi calls + call_history (list[QueueBlueapiCall]): List of completed blueapi calls Returns: list[BlueapiCall]: List of blueapi calls to execute. diff --git a/src/daq_queuing_service/plugins/i15_1/i15_1_converter.py b/src/daq_queuing_service/plugins/i15_1/i15_1_converter.py index 2a13a0f7..ef939492 100644 --- a/src/daq_queuing_service/plugins/i15_1/i15_1_converter.py +++ b/src/daq_queuing_service/plugins/i15_1/i15_1_converter.py @@ -7,6 +7,7 @@ from tiled.client.container import Container as TiledContainer from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( + BlueapiCall, QueueBlueapiCall, ) from daq_queuing_service.external_interaction.tiled.tiled import get_tiled_client @@ -98,23 +99,23 @@ def construct_blueapi_calls( self, queue: list[QueueTask], history: list[QueueTask], - call_history: list[QueueBlueapiCall], - ) -> list[QueueBlueapiCall]: - call_list: list[QueueBlueapiCall] = [] + call_history: list[BlueapiCall], + ) -> list[BlueapiCall]: + call_list: list[BlueapiCall] = [] for task in queue: with add_session_to_logger(task.experiment.instrument_session): match task.experiment: case TaskRequest(): call_list.append( - QueueBlueapiCall( + BlueapiCall( task_request=task.experiment, parent_task_id=task.id ) ) case Experiment(): call_list.extend( [ - QueueBlueapiCall( + BlueapiCall( task_request=b_api_task, parent_task_id=task.id ) for b_api_task in ( diff --git a/src/daq_queuing_service/task_queue/queue.py b/src/daq_queuing_service/task_queue/queue.py index 6de9aa84..8d307013 100644 --- a/src/daq_queuing_service/task_queue/queue.py +++ b/src/daq_queuing_service/task_queue/queue.py @@ -140,7 +140,6 @@ def __init__( self._call_queue: list[BlueapiCall] = [] self._call_history: list[BlueapiCall] = [] self._save_contents() - self._queue_history: list[BlueapiCall] = [] self._state: QueueState = QueueState( paused=True, last_pause_reason=PauseReason.EMPTY_QUEUE ) @@ -187,7 +186,7 @@ def _sync(self): if self._tasks[task_id].status == Status.QUEUED ], self._get_history(), - self._queue_history, + self._get_call_history(), ) except Exception as e: raise ConverterError(e) from e @@ -231,7 +230,7 @@ def _sync(self): if task.status == Status.QUEUED ], self._get_history(), - self._queue_history, + self._get_call_history(), ) ] except Exception as e: diff --git a/tests/unit_tests/test_queue.py b/tests/unit_tests/test_queue.py index af9c2229..feda8832 100644 --- a/tests/unit_tests/test_queue.py +++ b/tests/unit_tests/test_queue.py @@ -1214,7 +1214,7 @@ async def test_if_error_during_conversion_then_error_handled_and_contents_restor def convert( queue: list[QueueTask], history: list[QueueTask], - call_history: list[QueueBlueapiCall], + call_history: list[BlueapiCall], ): for task_id in task_queue._queue: del task_queue._tasks[task_id] @@ -1237,7 +1237,7 @@ async def test_if_error_during_conversion_then__restore_latest_good_contents_cal def convert( queue: list[QueueTask], history: list[QueueTask], - call_history: list[QueueBlueapiCall], + call_history: list[BlueapiCall], ): raise ValueError("Conversion failed") From 9e0646c24fccbf7f1c048fe3e49a6e834ae08f78 Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Tue, 3 Nov 2026 17:52:50 +0000 Subject: [PATCH 06/11] Always use QueueBlueapiCall in plugin --- src/daq_queuing_service/plugins/converter.py | 17 ++++++++--------- .../plugins/i15_1/i15_1_converter.py | 11 +++++------ tests/unit_tests/test_queue.py | 4 ++-- 3 files changed, 15 insertions(+), 17 deletions(-) diff --git a/src/daq_queuing_service/plugins/converter.py b/src/daq_queuing_service/plugins/converter.py index 6496f602..67090fb1 100644 --- a/src/daq_queuing_service/plugins/converter.py +++ b/src/daq_queuing_service/plugins/converter.py @@ -4,7 +4,6 @@ from blueapi.worker.event import TaskError, TaskResult from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( - BlueapiCall, QueueBlueapiCall, ) from daq_queuing_service.task_queue.task import ( @@ -74,11 +73,11 @@ def construct_blueapi_calls( queue: list[QueueTask], history: list[QueueTask], call_history: list[QueueBlueapiCall], - ) -> list[BlueapiCall]: + ) -> list[QueueBlueapiCall]: """Converts the list of queued tasks into a list of blueapi calls. This is needed if the queue contains ulims experiments, as these must be mapped onto bluesky plans. If the queue contains TaskRequests, they can just be - wrapped into BlueapiCall objects with no conversion, as this default + wrapped into QueueBlueapiCall objects with no conversion, as this default implementation does. Args: @@ -87,11 +86,11 @@ def construct_blueapi_calls( call_history (list[QueueBlueapiCall]): List of completed blueapi calls Returns: - list[BlueapiCall]: List of blueapi calls to execute. + list[QueueBlueapiCall]: List of blueapi calls to execute. """ call_list = [ - BlueapiCall( + QueueBlueapiCall( parent_task_id=task.id, task_request=self._construct_blueapi_task_request(task.experiment), ) @@ -100,12 +99,12 @@ def construct_blueapi_calls( return call_list def pause_on_complete_blueapi_call( - self, call: BlueapiCall, result: TaskResult + self, call: QueueBlueapiCall, result: TaskResult ) -> bool: """Decide whether to pause the queue after a blueapi call completes successfully Args: - call (BlueapiCall): The completed call + call (QueueBlueapiCall): The completed call result (TaskResult): The result of the call from blueapi Returns: @@ -114,12 +113,12 @@ def pause_on_complete_blueapi_call( return False def pause_on_fail_blueapi_call( - self, call: BlueapiCall, errors: list[str | TaskError] | None + self, call: QueueBlueapiCall, errors: list[str | TaskError] | None ) -> bool: """Decide whether to pause the queue after a blueapi call fails Args: - call (BlueapiCall): The failed call + call (QueueBlueapiCall): The failed call errors (list[str | TaskError] | None): The errors that occurred Returns: diff --git a/src/daq_queuing_service/plugins/i15_1/i15_1_converter.py b/src/daq_queuing_service/plugins/i15_1/i15_1_converter.py index ef939492..2a13a0f7 100644 --- a/src/daq_queuing_service/plugins/i15_1/i15_1_converter.py +++ b/src/daq_queuing_service/plugins/i15_1/i15_1_converter.py @@ -7,7 +7,6 @@ from tiled.client.container import Container as TiledContainer from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( - BlueapiCall, QueueBlueapiCall, ) from daq_queuing_service.external_interaction.tiled.tiled import get_tiled_client @@ -99,23 +98,23 @@ def construct_blueapi_calls( self, queue: list[QueueTask], history: list[QueueTask], - call_history: list[BlueapiCall], - ) -> list[BlueapiCall]: - call_list: list[BlueapiCall] = [] + call_history: list[QueueBlueapiCall], + ) -> list[QueueBlueapiCall]: + call_list: list[QueueBlueapiCall] = [] for task in queue: with add_session_to_logger(task.experiment.instrument_session): match task.experiment: case TaskRequest(): call_list.append( - BlueapiCall( + QueueBlueapiCall( task_request=task.experiment, parent_task_id=task.id ) ) case Experiment(): call_list.extend( [ - BlueapiCall( + QueueBlueapiCall( task_request=b_api_task, parent_task_id=task.id ) for b_api_task in ( diff --git a/tests/unit_tests/test_queue.py b/tests/unit_tests/test_queue.py index feda8832..af9c2229 100644 --- a/tests/unit_tests/test_queue.py +++ b/tests/unit_tests/test_queue.py @@ -1214,7 +1214,7 @@ async def test_if_error_during_conversion_then_error_handled_and_contents_restor def convert( queue: list[QueueTask], history: list[QueueTask], - call_history: list[BlueapiCall], + call_history: list[QueueBlueapiCall], ): for task_id in task_queue._queue: del task_queue._tasks[task_id] @@ -1237,7 +1237,7 @@ async def test_if_error_during_conversion_then__restore_latest_good_contents_cal def convert( queue: list[QueueTask], history: list[QueueTask], - call_history: list[BlueapiCall], + call_history: list[QueueBlueapiCall], ): raise ValueError("Conversion failed") From 8f2726f01b971dc15282b0b0d510756c8e81768d Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Tue, 3 Nov 2026 17:54:07 +0000 Subject: [PATCH 07/11] Fix lint --- tests/unit_tests/conftest.py | 10 +++++----- tests/unit_tests/test_api.py | 7 +------ 2 files changed, 6 insertions(+), 11 deletions(-) diff --git a/tests/unit_tests/conftest.py b/tests/unit_tests/conftest.py index 2846a673..742f3e4d 100644 --- a/tests/unit_tests/conftest.py +++ b/tests/unit_tests/conftest.py @@ -21,7 +21,7 @@ ) from daq_queuing_service.broadcaster import Broadcaster from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( - BlueapiCall, + QueueBlueapiCall, ) from daq_queuing_service.external_interaction.runtime_estimator.client import ( RuntimeEstimatorClient, @@ -144,14 +144,14 @@ async def task_queue_one_to_many(tasks: list[Task]): def construct_blueapi_calls( queue: list[QueueTask], history: list[QueueTask], - call_history: list[BlueapiCall], - ) -> list[BlueapiCall]: - call_list: list[BlueapiCall] = [] + call_history: list[QueueBlueapiCall], + ) -> list[QueueBlueapiCall]: + call_list: list[QueueBlueapiCall] = [] for task in queue: for _ in range(2): assert isinstance(task.experiment, Experiment) call_list.append( - BlueapiCall( + QueueBlueapiCall( parent_task_id=task.id, task_request=TaskRequest( name=task.experiment.experiment_definition.name, diff --git a/tests/unit_tests/test_api.py b/tests/unit_tests/test_api.py index 899757a9..1b659a8f 100644 --- a/tests/unit_tests/test_api.py +++ b/tests/unit_tests/test_api.py @@ -30,7 +30,6 @@ from daq_queuing_service.app.config import load_config from daq_queuing_service.broadcaster import Broadcaster, Event from daq_queuing_service.external_interaction.blueapi.blueapi_call import ( - BlueapiCall, CallStatus, QueueBlueapiCall, ) @@ -42,7 +41,6 @@ QueueTask, Status, Task, - TaskKind, ) @@ -432,14 +430,11 @@ async def test_add_tasks_to_queue_adds_to_queue_and_and_returns_task_ids( blueapi_id=None, tiled_ids=[], scan_ids=[], - runtime=None, ) ], position=3, status=Status.QUEUED, - kind=TaskKind.PLAN, user=None, - runtime=None, estimated_runtime=100, ) @@ -499,7 +494,7 @@ def fail_conversion( current_task: QueueTask | None, queue: list[Task], history: list[QueueTask], - call_history: list[BlueapiCall], + call_history: list[QueueBlueapiCall], ): raise SomeError("Conversion failed because xyz") From 5c00ce121e125503c9553b4732f372d32e3b598f Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Tue, 3 Nov 2026 18:12:29 +0000 Subject: [PATCH 08/11] Add docstrings --- .../external_interaction/blueapi/blueapi_call.py | 8 ++++++++ src/daq_queuing_service/task_queue/task.py | 14 ++++++++++++++ 2 files changed, 22 insertions(+) diff --git a/src/daq_queuing_service/external_interaction/blueapi/blueapi_call.py b/src/daq_queuing_service/external_interaction/blueapi/blueapi_call.py index c1638204..ca321c91 100644 --- a/src/daq_queuing_service/external_interaction/blueapi/blueapi_call.py +++ b/src/daq_queuing_service/external_interaction/blueapi/blueapi_call.py @@ -32,6 +32,10 @@ def allowed_transitions(self): class QueueBlueapiCall(BaseModel): + """A class representing a call to blueapi, containing the plan parameters and + some associated metadata. Constructed by a queue plugin. + """ + task_request: TaskRequest parent_task_id: str status: CallStatus = CallStatus.WAITING @@ -51,6 +55,10 @@ def runtime(self) -> float | None: class BlueapiCall(QueueBlueapiCall): + """A QueueBlueapiCall for internal use inside the queue, with added methods for + state management. + """ + def _update_status(self, new_status: CallStatus): """Updates the status of the task, checking that the transition is valid diff --git a/src/daq_queuing_service/task_queue/task.py b/src/daq_queuing_service/task_queue/task.py index e49c8087..b3a142ad 100644 --- a/src/daq_queuing_service/task_queue/task.py +++ b/src/daq_queuing_service/task_queue/task.py @@ -64,6 +64,16 @@ class TaskKind(StrEnum): class BaseTask(BaseModel, Generic[T]): + """Base class for a queue task. Contains either a ulims experiment or blueapi task + request, and a list of blueapi calls to be made. + + Args: + + Generic (QueueBlueapiCall | BlueapiCall): The type of BlueapiCall contained in + the blueapi_calls field. This is to differentiate between objects used by the + queue internally and ones copied to pass to queue plugins or the API layer. + """ + experiment: Experiment | TaskRequest id: str = Field(default_factory=create_uuid_str, frozen=True) blueapi_calls: list[T] = Field(default_factory=lambda: []) @@ -94,6 +104,8 @@ def runtime(self) -> float | None: class Task(BaseTask[BlueapiCall]): + """Internal task object held by the queue.""" + _cancelled: bool = PrivateAttr(False) def cancel(self): @@ -124,6 +136,8 @@ def status(self) -> Status: class QueueTask(BaseTask[QueueBlueapiCall]): + """Task object with a position field, for passing to a queue plugin or API layer.""" + position: int | None status: Status From 51e827fdccd48814fa5c2825a64b6e6ceb986105 Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Tue, 3 Nov 2026 18:16:46 +0000 Subject: [PATCH 09/11] Update docs --- docs/explanations/converters.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/docs/explanations/converters.md b/docs/explanations/converters.md index 06ed4f5d..dfcea677 100644 --- a/docs/explanations/converters.md +++ b/docs/explanations/converters.md @@ -23,15 +23,15 @@ class MyConverter(Converter): self, queue: list[Task], history: list[QueueTask], - call_history: list[BlueapiCall], + call_history: list[QueueBlueapiCall], ) -> list[Task]: ... def construct_blueapi_calls( self, queue: list[QueueTask], history: list[QueueTask], - call_history: list[BlueapiCall], - ) -> list[BlueapiCall]: ... + call_history: list[QueueBlueapiCall], + ) -> list[QueueBlueapiCall]: ... ``` ### `validate()` @@ -44,7 +44,7 @@ This is called whenever anything changes in the queue (e.g a task is added, or a ### `construct_blueapi_calls()` -The queue contains ulims experiments and/or TaskRequests. While TaskRequests are already in the right format for blueapi to receive them, experiments need to be converted to blueapi calls for anything to run. This method is used for this conversion, constructing a list of BlueapiCall objects (which wrap TaskRequest objects) from the current queue contents. This is run whenever the queue changes, but after `pre_process`. The default converter assumes that the queue contains exclusively TaskRequests, and will raise a `NotImplementedError` if an `Experiment` is added. If you want to add ulims experiments to the queue, make sure you are using a custom converter that overrides this method! +The queue contains ulims experiments and/or TaskRequests. While TaskRequests are already in the right format for blueapi to receive them, experiments need to be converted to blueapi calls for anything to run. This method is used for this conversion, constructing a list of QueueBlueapiCall objects (which wrap TaskRequest objects) from the current queue contents. This is run whenever the queue changes, but after `pre_process`. The default converter assumes that the queue contains exclusively TaskRequests, and will raise a `NotImplementedError` if an `Experiment` is added. If you want to add ulims experiments to the queue, make sure you are using a custom converter that overrides this method! Pre-processing and conversion is run any time the queue changes and right before a call is made to blueapi, so the blueapi call list stays in sync. This means these methods can make use of current information on a beamline such as PV values to inform their logic. However, because they're run so frequently, it is important these methods do not take a large amount of time. From cfd570257a935280e796a749d58ad696bbf954f3 Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Wed, 4 Nov 2026 17:16:33 +0000 Subject: [PATCH 10/11] Remove line --- src/daq_queuing_service/task_queue/task.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/daq_queuing_service/task_queue/task.py b/src/daq_queuing_service/task_queue/task.py index b3a142ad..af496316 100644 --- a/src/daq_queuing_service/task_queue/task.py +++ b/src/daq_queuing_service/task_queue/task.py @@ -68,7 +68,6 @@ class BaseTask(BaseModel, Generic[T]): request, and a list of blueapi calls to be made. Args: - Generic (QueueBlueapiCall | BlueapiCall): The type of BlueapiCall contained in the blueapi_calls field. This is to differentiate between objects used by the queue internally and ones copied to pass to queue plugins or the API layer. From 5fd4d9bb7c99500a76c6118b4baac5cf3aea2984 Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Wed, 4 Nov 2026 17:18:16 +0000 Subject: [PATCH 11/11] Update api docs --- docs/reference/rest_api.json | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/docs/reference/rest_api.json b/docs/reference/rest_api.json index a7cde4e8..3aa33b58 100644 --- a/docs/reference/rest_api.json +++ b/docs/reference/rest_api.json @@ -1097,7 +1097,8 @@ "parent_task_id", "runtime" ], - "title": "QueueBlueapiCall" + "title": "QueueBlueapiCall", + "description": "A class representing a call to blueapi, containing the plan parameters and\nsome associated metadata. Constructed by a queue plugin." }, "QueueState": { "properties": { @@ -1224,7 +1225,8 @@ "kind", "runtime" ], - "title": "QueueTask" + "title": "QueueTask", + "description": "Task object with a position field, for passing to a queue plugin or API layer." }, "RestConfig": { "properties": {