From 8c63ea320c06192b1c6f74f8bdf07fa864cc3230 Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Fri, 12 Jun 2026 14:52:26 +0100 Subject: [PATCH 01/10] Start on client --- src/daq_queuing_service/client/client.py | 79 ++++++++++++++++++++++++ 1 file changed, 79 insertions(+) create mode 100644 src/daq_queuing_service/client/client.py diff --git a/src/daq_queuing_service/client/client.py b/src/daq_queuing_service/client/client.py new file mode 100644 index 0000000..98201b2 --- /dev/null +++ b/src/daq_queuing_service/client/client.py @@ -0,0 +1,79 @@ +from collections.abc import Mapping +from typing import Any, TypeVar + +import requests +from pydantic import HttpUrl, TypeAdapter + +from daq_queuing_service.api.api import TaskCancelRequest +from daq_queuing_service.task import ExperimentDefinition, TaskWithPosition +from daq_queuing_service.task_queue.queue import QueueState + +T = TypeVar("T") + + +class QueueClient: + def __init__(self, url: str): + self._url = HttpUrl(url) + self._pool = requests.Session() + + def _request( + self, + suffix: str, + target_type: type[T], + method: str = "GET", + data: Any = None, + params: Mapping[str, Any] | None = None, + ): + url = self._url.unicode_string().removesuffix("/") + suffix + response = self._pool.request( + method, + url, + json=data, + params=params, + ) + + return TypeAdapter(target_type).validate_python(response.json()) + + def get_queue_state(self) -> QueueState: + return self._request("/queue/state", QueueState) + + def update_queue_state(self, new_state: QueueState) -> QueueState: + return self._request( + "/queue/state", QueueState, method="PATCH", data=new_state.model_dump() + ) + + def get_queued_tasks(self) -> list[TaskWithPosition]: + return self._request("/queue", list[TaskWithPosition]) + + def add_tasks_to_queue( + self, + experiment_definitions: list[ExperimentDefinition], + position: int | None = None, + ) -> list[str]: + return self._request( + "/queue", + list[str], + method="POST", + data=[exp_def.model_dump() for exp_def in experiment_definitions], + params={"position": position}, + ) + + def move_task(self, task_id: str, new_position: int): + return self._request( + "/queue/move", + int, + method="POST", + params={"task_id": task_id, "new_position": new_position}, + ) + + def cancel_tasks(self, task_ids: list[str]) -> list[TaskWithPosition]: + return self._request( + "/queue/tasks", + list[TaskWithPosition], + method="DELETE", + data=TaskCancelRequest(task_ids=task_ids).model_dump(), + ) + + def cancel_all_tasks(self) -> list[TaskWithPosition]: + task_ids = [task.id for task in self.get_queued_tasks()] + return self.cancel_tasks(task_ids) From d8b0f9f71ee8c052bfc8bfb40bf34c6761ed1cce Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Fri, 12 Jun 2026 15:57:40 +0100 Subject: [PATCH 02/10] Add no validation option to client --- src/daq_queuing_service/client/client.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/daq_queuing_service/client/client.py b/src/daq_queuing_service/client/client.py index 98201b2..bd456f3 100644 --- a/src/daq_queuing_service/client/client.py +++ b/src/daq_queuing_service/client/client.py @@ -49,13 +49,17 @@ def add_tasks_to_queue( self, experiment_definitions: list[ExperimentDefinition], position: int | None = None, + validate_with_blueapi: bool = True, ) -> list[str]: return self._request( "/queue", list[str], method="POST", data=[exp_def.model_dump() for exp_def in experiment_definitions], - params={"position": position}, + params={ + "position": position, + "validate_with_blueapi": validate_with_blueapi, + }, ) def move_task(self, task_id: str, new_position: int): From f31ab70a873d556aeaf5efa08aa4c576a6bfb66d Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Fri, 12 Jun 2026 16:18:33 +0100 Subject: [PATCH 03/10] Add comment --- src/daq_queuing_service/client/client.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/daq_queuing_service/client/client.py b/src/daq_queuing_service/client/client.py index bd456f3..bc3320a 100644 --- a/src/daq_queuing_service/client/client.py +++ b/src/daq_queuing_service/client/client.py @@ -78,6 +78,8 @@ def cancel_tasks(self, task_ids: list[str]) -> list[TaskWithPosition]: data=TaskCancelRequest(task_ids=task_ids).model_dump(), ) + # TODO: Should add an endpoint for this. + # Potential race condition between the GET and DELETE def cancel_all_tasks(self) -> list[TaskWithPosition]: task_ids = [task.id for task in self.get_queued_tasks()] return self.cancel_tasks(task_ids) From 376ed34164225810f7a96fdfbeaed93c6d55eda1 Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Fri, 12 Jun 2026 16:20:04 +0100 Subject: [PATCH 04/10] Add script to take many scans on i15-1 --- src/add_1000_tasks_script.py | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) create mode 100644 src/add_1000_tasks_script.py diff --git a/src/add_1000_tasks_script.py b/src/add_1000_tasks_script.py new file mode 100644 index 0000000..16ecfcd --- /dev/null +++ b/src/add_1000_tasks_script.py @@ -0,0 +1,26 @@ +from daq_queuing_service.client.client import QueueClient +from daq_queuing_service.task import ExperimentDefinition + + +def main(): + client = QueueClient("https://i15-1-daq-queue.diamond.ac.uk/") + n_scans = 1000 + + # client.cancel_all_tasks() + + client.add_tasks_to_queue( + [ + ExperimentDefinition( + plan_name="static_collect_and_trigger_analysis", + sample_id=f"COLLECTION {i}", + params={"frames": 200, "exposure_time": 0.05}, + instrument_session="cm44163-3", + ) + for i in range(n_scans) + ], + validate_with_blueapi=False, + )[0] + + +if __name__ == "__main__": + main() From 270304bae64b1d3382ec228801a8350f6a79f794 Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Tue, 7 Jul 2026 09:37:11 +0100 Subject: [PATCH 05/10] Add more client functions --- src/daq_queuing_service/api/api.py | 4 +-- src/daq_queuing_service/client/client.py | 34 +++++++++++++++++++++--- src/daq_queuing_service/worker/worker.py | 5 ++-- tests/unit_tests/client/test_client.py | 21 +++++++++++++++ 4 files changed, 55 insertions(+), 9 deletions(-) create mode 100644 tests/unit_tests/client/test_client.py diff --git a/src/daq_queuing_service/api/api.py b/src/daq_queuing_service/api/api.py index da3aec3..efdeabf 100644 --- a/src/daq_queuing_service/api/api.py +++ b/src/daq_queuing_service/api/api.py @@ -45,11 +45,11 @@ def create_api_router( router = APIRouter() @router.get("/healthz") - async def healthz(): + async def healthz() -> Response: return Response() @router.get("/") - def read_root(request: Request): + def read_root(request: Request) -> str: base_url = str(request.base_url) return ( f"Welcome to the daq queuing service. Visit {base_url}docs for Uvicorn API." diff --git a/src/daq_queuing_service/client/client.py b/src/daq_queuing_service/client/client.py index bc3320a..9c6c719 100644 --- a/src/daq_queuing_service/client/client.py +++ b/src/daq_queuing_service/client/client.py @@ -5,6 +5,8 @@ from pydantic import HttpUrl, TypeAdapter from daq_queuing_service.api.api import TaskCancelRequest +from daq_queuing_service.app._config import AppConfig +from daq_queuing_service.blueapi_interaction.blueapi_call import BlueapiCallResponse from daq_queuing_service.task import ExperimentDefinition, TaskWithPosition from daq_queuing_service.task_queue.queue import QueueState @@ -34,6 +36,12 @@ def _request( return TypeAdapter(target_type).validate_python(response.json()) + def healthz(self): + return self._request("/healthz", str) + + def get_config(self): + return self._request("/config", AppConfig) + def get_queue_state(self) -> QueueState: return self._request("/queue/state", QueueState) @@ -78,8 +86,26 @@ def cancel_tasks(self, task_ids: list[str]) -> list[TaskWithPosition]: data=TaskCancelRequest(task_ids=task_ids).model_dump(), ) - # TODO: Should add an endpoint for this. - # Potential race condition between the GET and DELETE def cancel_all_tasks(self) -> list[TaskWithPosition]: - task_ids = [task.id for task in self.get_queued_tasks()] - return self.cancel_tasks(task_ids) + return self.cancel_all_tasks() + + def get_task_by_position(self, position: int) -> TaskWithPosition: + return self._request(f"/queue/{position}", TaskWithPosition) + + def get_all_tasks(self) -> list[TaskWithPosition]: + return self._request("/tasks", list[TaskWithPosition]) + + def get_task_by_id(self, task_id: str) -> TaskWithPosition: + return self._request(f"/tasks/{task_id}", TaskWithPosition) + + def get_completed_tasks(self) -> list[TaskWithPosition]: + return self._request("/history", list[TaskWithPosition]) + + def clear_history(self): + return self._request("/history", str, "DELETE") + + def get_call_queue(self) -> list[BlueapiCallResponse]: + return self._request("/call_queue", list[BlueapiCallResponse]) + + def get_call_history(self) -> list[BlueapiCallResponse]: + return self._request("/call_history", list[BlueapiCallResponse]) diff --git a/src/daq_queuing_service/worker/worker.py b/src/daq_queuing_service/worker/worker.py index 11ed2bf..bae0260 100644 --- a/src/daq_queuing_service/worker/worker.py +++ b/src/daq_queuing_service/worker/worker.py @@ -80,7 +80,7 @@ async def _process_call(self, call: BlueapiCall): match task_status.result: case TaskResult(): LOGGER.debug( - f"Call {call} completed succesfully: {task_status.result}" + f"Call {call} completed successfully: {task_status.result}" ) await self._queue.complete_call(call, task_status.result) case TaskError(): @@ -115,8 +115,7 @@ async def _handle_run_task_error( error: InvalidParametersError | UnknownPlanError | ServiceUnavailableError - | BlueskyRemoteControlError - | ServiceUnavailableError, + | BlueskyRemoteControlError, ): match error: case InvalidParametersError(): diff --git a/tests/unit_tests/client/test_client.py b/tests/unit_tests/client/test_client.py new file mode 100644 index 0000000..ba12823 --- /dev/null +++ b/tests/unit_tests/client/test_client.py @@ -0,0 +1,21 @@ +from unittest.mock import MagicMock + +from daq_queuing_service.api.api import create_api_router +from daq_queuing_service.client.client import QueueClient + + +def test_all_endpoints_have_a_client_function(): + exceptions = ["read_root", "stream_events"] + + router = create_api_router(MagicMock(), MagicMock(), MagicMock()) + endpoints: list[str] = [route.endpoint.__name__ for route in router.routes] # type: ignore + + not_in_client = [ + endpoint + for endpoint in endpoints + if endpoint not in list(QueueClient.__dict__.keys()) + exceptions + ] + + assert not not_in_client, ( + f"Found endpoints not covered by a client function: {not_in_client}" + ) From 970ca45e333d91af90ea55874cb30d5b560a4c97 Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Tue, 7 Jul 2026 10:55:15 +0100 Subject: [PATCH 06/10] Deal with different return types --- src/daq_queuing_service/api/errors.py | 25 ++++-- .../blueapi_interaction/clients.py | 5 -- src/daq_queuing_service/client/client.py | 85 ++++++++++++++----- 3 files changed, 82 insertions(+), 33 deletions(-) diff --git a/src/daq_queuing_service/api/errors.py b/src/daq_queuing_service/api/errors.py index ddebaf0..93b249b 100644 --- a/src/daq_queuing_service/api/errors.py +++ b/src/daq_queuing_service/api/errors.py @@ -1,5 +1,6 @@ from fastapi import FastAPI, Request from fastapi.responses import JSONResponse +from pydantic import BaseModel from daq_queuing_service.task_queue.queue_utils import ( NegativePositionError, @@ -9,7 +10,11 @@ TaskNotInQueueError, ) + # pyright: reportUnusedFunction=false +class ErrorContent(BaseModel): + error: str + message: str def register_exception_handlers(app: FastAPI): @@ -19,14 +24,18 @@ async def task_in_progress_handler( ): return JSONResponse( status_code=409, - content={"error": "task_in_progress", "message": str(exception)}, + content=ErrorContent( + error="task_in_progress", message=str(exception) + ).model_dump(), ) @app.exception_handler(TaskNotFoundError) async def task_not_found_handler(request: Request, exception: TaskNotFoundError): return JSONResponse( status_code=404, - content={"error": "task_not_found", "message": str(exception)}, + content=ErrorContent( + error="task_not_found", message=str(exception) + ).model_dump(), ) @app.exception_handler(TaskNotInQueueError) @@ -35,7 +44,9 @@ async def task_not_in_queue_handler( ): return JSONResponse( status_code=409, - content={"error": "task_not_in_queue", "message": str(exception)}, + content=ErrorContent( + error="task_not_in_queue", message=str(exception) + ).model_dump(), ) @app.exception_handler(NegativePositionError) @@ -44,12 +55,16 @@ async def negative_position_handler( ): return JSONResponse( status_code=400, - content={"error": "negative_position", "message": str(exception)}, + content=ErrorContent( + error="negative_position", message=str(exception) + ).model_dump(), ) @app.exception_handler(QueueError) async def queue_error_handler(request: Request, exception: QueueError): return JSONResponse( status_code=409, - content={"error": "queue_error", "message": str(exception)}, + content=ErrorContent( + error="queue_error", message=str(exception) + ).model_dump(), ) diff --git a/src/daq_queuing_service/blueapi_interaction/clients.py b/src/daq_queuing_service/blueapi_interaction/clients.py index 3591191..7677441 100644 --- a/src/daq_queuing_service/blueapi_interaction/clients.py +++ b/src/daq_queuing_service/blueapi_interaction/clients.py @@ -1,5 +1,3 @@ -from unittest.mock import MagicMock - from blueapi.client import BlueapiClient from blueapi.client.event_bus import EventBusClient from blueapi.client.rest import BlueapiRestClient @@ -10,9 +8,6 @@ def get_blueapi_clients(blueapi_config: ApplicationConfig) -> BlueapiClient: - if not blueapi_config.oidc: - blueapi_config.oidc = MagicMock() - blueapi_rest_client = BlueapiRestClient( config=blueapi_config.api, # Waiting on https://github.com/DiamondLightSource/blueapi/pull/1553 diff --git a/src/daq_queuing_service/client/client.py b/src/daq_queuing_service/client/client.py index 9c6c719..43aa38d 100644 --- a/src/daq_queuing_service/client/client.py +++ b/src/daq_queuing_service/client/client.py @@ -2,9 +2,10 @@ from typing import Any, TypeVar import requests -from pydantic import HttpUrl, TypeAdapter +from pydantic import HttpUrl, TypeAdapter, ValidationError from daq_queuing_service.api.api import TaskCancelRequest +from daq_queuing_service.api.errors import ErrorContent from daq_queuing_service.app._config import AppConfig from daq_queuing_service.blueapi_interaction.blueapi_call import BlueapiCallResponse from daq_queuing_service.task import ExperimentDefinition, TaskWithPosition @@ -21,7 +22,6 @@ def __init__(self, url: str): def _request( self, suffix: str, - target_type: type[T], method: str = "GET", data: Any = None, params: Mapping[str, Any] | None = None, @@ -33,33 +33,70 @@ def _request( json=data, params=params, ) + return response + + def _request_expect_error( + self, + suffix: str, + target_type: type[T], + method: str = "GET", + data: Any = None, + params: Mapping[str, Any] | None = None, + ): + response = self._request(suffix, method, data, params) + try: + return TypeAdapter(target_type).validate_python(response.json()) + except ValidationError: + return ErrorContent.model_validate(response.json()) + def _request_expect_none( + self, + suffix: str, + target_type: type[T], + method: str = "GET", + data: Any = None, + params: Mapping[str, Any] | None = None, + ) -> T | None: + response = self._request(suffix, method, data, params) + if response.json() is None: + return + return TypeAdapter(target_type).validate_python(response.json()) + + def _request_and_validate( + self, + suffix: str, + target_type: type[T], + method: str = "GET", + data: Any = None, + params: Mapping[str, Any] | None = None, + ) -> T: + response = self._request(suffix, method, data, params) return TypeAdapter(target_type).validate_python(response.json()) def healthz(self): - return self._request("/healthz", str) + return self._request_and_validate("/healthz", str) def get_config(self): - return self._request("/config", AppConfig) + return self._request_and_validate("/config", AppConfig) def get_queue_state(self) -> QueueState: - return self._request("/queue/state", QueueState) + return self._request_and_validate("/queue/state", QueueState) def update_queue_state(self, new_state: QueueState) -> QueueState: - return self._request( + return self._request_and_validate( "/queue/state", QueueState, method="PATCH", data=new_state.model_dump() ) def get_queued_tasks(self) -> list[TaskWithPosition]: - return self._request("/queue", list[TaskWithPosition]) + return self._request_and_validate("/queue", list[TaskWithPosition]) def add_tasks_to_queue( self, experiment_definitions: list[ExperimentDefinition], position: int | None = None, validate_with_blueapi: bool = True, - ) -> list[str]: - return self._request( + ) -> list[str] | ErrorContent: + return self._request_expect_error( "/queue", list[str], method="POST", @@ -70,16 +107,18 @@ def add_tasks_to_queue( }, ) - def move_task(self, task_id: str, new_position: int): - return self._request( + def move_task(self, task_id: str, new_position: int) -> int | ErrorContent: + return self._request_expect_error( "/queue/move", int, method="POST", params={"task_id": task_id, "new_position": new_position}, ) - def cancel_tasks(self, task_ids: list[str]) -> list[TaskWithPosition]: - return self._request( + def cancel_tasks( + self, task_ids: list[str] + ) -> list[TaskWithPosition] | ErrorContent: + return self._request_expect_error( "/queue/tasks", list[TaskWithPosition], method="DELETE", @@ -87,25 +126,25 @@ def cancel_tasks(self, task_ids: list[str]) -> list[TaskWithPosition]: ) def cancel_all_tasks(self) -> list[TaskWithPosition]: - return self.cancel_all_tasks() + return self._request_and_validate("/queue", list[TaskWithPosition], "DELETE") - def get_task_by_position(self, position: int) -> TaskWithPosition: - return self._request(f"/queue/{position}", TaskWithPosition) + def get_task_by_position(self, position: int) -> TaskWithPosition | None: + return self._request_expect_none(f"/queue/{position}", TaskWithPosition) def get_all_tasks(self) -> list[TaskWithPosition]: - return self._request("/tasks", list[TaskWithPosition]) + return self._request_and_validate("/tasks", list[TaskWithPosition]) - def get_task_by_id(self, task_id: str) -> TaskWithPosition: - return self._request(f"/tasks/{task_id}", TaskWithPosition) + def get_task_by_id(self, task_id: str) -> TaskWithPosition | ErrorContent: + return self._request_expect_error(f"/tasks/{task_id}", TaskWithPosition) def get_completed_tasks(self) -> list[TaskWithPosition]: - return self._request("/history", list[TaskWithPosition]) + return self._request_and_validate("/history", list[TaskWithPosition]) def clear_history(self): - return self._request("/history", str, "DELETE") + return self._request_and_validate("/history", str, "DELETE") def get_call_queue(self) -> list[BlueapiCallResponse]: - return self._request("/call_queue", list[BlueapiCallResponse]) + return self._request_and_validate("/call_queue", list[BlueapiCallResponse]) def get_call_history(self) -> list[BlueapiCallResponse]: - return self._request("/call_history", list[BlueapiCallResponse]) + return self._request_and_validate("/call_history", list[BlueapiCallResponse]) From be87848c7d35dbd91d6a025662a3be1df46e7962 Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Tue, 7 Jul 2026 11:24:12 +0100 Subject: [PATCH 07/10] Fixes --- src/daq_queuing_service/client/client.py | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/src/daq_queuing_service/client/client.py b/src/daq_queuing_service/client/client.py index 43aa38d..a69c1c1 100644 --- a/src/daq_queuing_service/client/client.py +++ b/src/daq_queuing_service/client/client.py @@ -2,13 +2,14 @@ from typing import Any, TypeVar import requests +from blueapi.service.model import TaskRequest from pydantic import HttpUrl, TypeAdapter, ValidationError from daq_queuing_service.api.api import TaskCancelRequest from daq_queuing_service.api.errors import ErrorContent from daq_queuing_service.app._config import AppConfig from daq_queuing_service.blueapi_interaction.blueapi_call import BlueapiCallResponse -from daq_queuing_service.task import ExperimentDefinition, TaskWithPosition +from daq_queuing_service.task import Experiment, TaskWithPosition from daq_queuing_service.task_queue.queue import QueueState T = TypeVar("T") @@ -92,19 +93,15 @@ def get_queued_tasks(self) -> list[TaskWithPosition]: def add_tasks_to_queue( self, - experiment_definitions: list[ExperimentDefinition], + experiments: list[Experiment | TaskRequest], position: int | None = None, - validate_with_blueapi: bool = True, ) -> list[str] | ErrorContent: return self._request_expect_error( "/queue", list[str], method="POST", - data=[exp_def.model_dump() for exp_def in experiment_definitions], - params={ - "position": position, - "validate_with_blueapi": validate_with_blueapi, - }, + data=[experiment.model_dump() for experiment in experiments], + params={"position": position}, ) def move_task(self, task_id: str, new_position: int) -> int | ErrorContent: From 5a2ddade2de785801f86f93b7b09de3dcd0527a0 Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Tue, 7 Jul 2026 11:25:32 +0100 Subject: [PATCH 08/10] Move script into docs --- docs/examples/add_tasks_using_client.py | 24 +++++++++++++++++++++++ src/add_1000_tasks_script.py | 26 ------------------------- 2 files changed, 24 insertions(+), 26 deletions(-) create mode 100644 docs/examples/add_tasks_using_client.py delete mode 100644 src/add_1000_tasks_script.py diff --git a/docs/examples/add_tasks_using_client.py b/docs/examples/add_tasks_using_client.py new file mode 100644 index 0000000..c65cb8c --- /dev/null +++ b/docs/examples/add_tasks_using_client.py @@ -0,0 +1,24 @@ +from blueapi.service.model import TaskRequest + +from daq_queuing_service.client.client import QueueClient + +# Example of how to add tasks to the queue using the Python client +# Need a local queue running on port 8001 for this to work + + +def main(): + client = QueueClient("http://127.0.0.1:8001") + n_tasks = 10 + + client.add_tasks_to_queue( + [ + TaskRequest( + name="sleep", instrument_session="cm44163-3", params={"time": 10} + ) + for _ in range(n_tasks) + ], + ) + + +if __name__ == "__main__": + main() diff --git a/src/add_1000_tasks_script.py b/src/add_1000_tasks_script.py deleted file mode 100644 index 16ecfcd..0000000 --- a/src/add_1000_tasks_script.py +++ /dev/null @@ -1,26 +0,0 @@ -from daq_queuing_service.client.client import QueueClient -from daq_queuing_service.task import ExperimentDefinition - - -def main(): - client = QueueClient("https://i15-1-daq-queue.diamond.ac.uk/") - n_scans = 1000 - - # client.cancel_all_tasks() - - client.add_tasks_to_queue( - [ - ExperimentDefinition( - plan_name="static_collect_and_trigger_analysis", - sample_id=f"COLLECTION {i}", - params={"frames": 200, "exposure_time": 0.05}, - instrument_session="cm44163-3", - ) - for i in range(n_scans) - ], - validate_with_blueapi=False, - )[0] - - -if __name__ == "__main__": - main() From 6709404dbb1be905455917c74014d014508e5e0e Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Tue, 7 Jul 2026 11:37:11 +0100 Subject: [PATCH 09/10] Improve example --- docs/examples/add_tasks_using_client.py | 28 ++++++++++++++++++++++--- 1 file changed, 25 insertions(+), 3 deletions(-) diff --git a/docs/examples/add_tasks_using_client.py b/docs/examples/add_tasks_using_client.py index c65cb8c..e62c700 100644 --- a/docs/examples/add_tasks_using_client.py +++ b/docs/examples/add_tasks_using_client.py @@ -1,6 +1,7 @@ from blueapi.service.model import TaskRequest from daq_queuing_service.client.client import QueueClient +from daq_queuing_service.task_queue.queue import QueueState # Example of how to add tasks to the queue using the Python client # Need a local queue running on port 8001 for this to work @@ -8,17 +9,38 @@ def main(): client = QueueClient("http://127.0.0.1:8001") - n_tasks = 10 + # Pause the queue + client.update_queue_state(QueueState(paused=True)) + + # Clear everything in the queue + client.cancel_all_tasks() + + # Add 10 sleep plans to the queue client.add_tasks_to_queue( [ TaskRequest( - name="sleep", instrument_session="cm44163-3", params={"time": 10} + name="sleep", instrument_session="cm44163-3", params={"time": 1} ) - for _ in range(n_tasks) + for _ in range(10) ], ) + # Add a longer sleep to the start of the queue + client.add_tasks_to_queue( + [TaskRequest(name="sleep", instrument_session="cm44163-3", params={"time": 5})], + position=0, + ) + + # Print the current tasks in the queue + tasks = client.get_queued_tasks() + for task in tasks: + assert isinstance(task.experiment, TaskRequest) + print( + f"Task {task.position}: {task.experiment.name} for " + + f"{task.experiment.params['time']} seconds" + ) + if __name__ == "__main__": main() From 799950257d9fd7da959efa35d1f8b165112ef524 Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Tue, 7 Jul 2026 12:07:01 +0100 Subject: [PATCH 10/10] Add a test --- tests/unit_tests/client/test_client.py | 44 +++++++++++++++++++++++++- 1 file changed, 43 insertions(+), 1 deletion(-) diff --git a/tests/unit_tests/client/test_client.py b/tests/unit_tests/client/test_client.py index ba12823..72e0785 100644 --- a/tests/unit_tests/client/test_client.py +++ b/tests/unit_tests/client/test_client.py @@ -1,4 +1,7 @@ -from unittest.mock import MagicMock +from typing import Any +from unittest.mock import MagicMock, _Call, call, patch + +import pytest from daq_queuing_service.api.api import create_api_router from daq_queuing_service.client.client import QueueClient @@ -19,3 +22,42 @@ def test_all_endpoints_have_a_client_function(): assert not not_in_client, ( f"Found endpoints not covered by a client function: {not_in_client}" ) + + +@pytest.mark.parametrize( + "url, kwargs, expected_request_call", + [ + ( + "https://google.com/", + {"suffix": "/queue"}, + call("GET", "https://google.com/queue", json=None, params=None), + ), + ( + "http://google.com", + { + "suffix": "/queue", + "method": "POST", + "data": {"test": "data"}, + "params": {"test": "params"}, + }, + call( + "POST", + "http://google.com/queue", + json={"test": "data"}, + params={"test": "params"}, + ), + ), + ], +) +def test__request_makes_request_with_expected_arguments( + url: str, kwargs: dict[str, Any], expected_request_call: _Call +): + client = QueueClient(url=url) + + with patch( + "daq_queuing_service.client.client.requests.Session.request" + ) as mock_request: + client._request(**kwargs) + + mock_request.assert_called_once() + mock_request.assert_has_calls([expected_request_call])