From ff91bf55622a8c75362bfe29af606ac458493d6f Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 9 Sep 2026 15:01:37 +0000 Subject: [PATCH 01/31] Support *args in plan args --- src/blueapi/worker/__init__.py | 3 +- src/blueapi/worker/task.py | 87 +++++++++++++++------ tests/unit_tests/service/test_rest_api.py | 17 ++-- tests/unit_tests/worker/test_task_worker.py | 54 +++++++------ 4 files changed, 106 insertions(+), 55 deletions(-) diff --git a/src/blueapi/worker/__init__.py b/src/blueapi/worker/__init__.py index 85ae49b453..5002da8a8c 100644 --- a/src/blueapi/worker/__init__.py +++ b/src/blueapi/worker/__init__.py @@ -1,11 +1,12 @@ from .event import ProgressEvent, StatusView, TaskStatus, WorkerEvent, WorkerState -from .task import Task +from .task import Task, TaskParams from .task_worker import TaskWorker, TrackableTask from .worker_errors import WorkerAlreadyStartedError, WorkerBusyError __all__ = [ "TaskWorker", "Task", + "TaskParams", "WorkerEvent", "WorkerState", "StatusView", diff --git a/src/blueapi/worker/task.py b/src/blueapi/worker/task.py index e2efade0d7..648f442947 100644 --- a/src/blueapi/worker/task.py +++ b/src/blueapi/worker/task.py @@ -1,8 +1,9 @@ import logging from collections.abc import Mapping +from inspect import Parameter, signature from typing import Any -from pydantic import BaseModel, Field, TypeAdapter +from pydantic import Field, TypeAdapter from blueapi.core import BlueskyContext from blueapi.utils import BlueapiBaseModel @@ -10,55 +11,93 @@ LOGGER = logging.getLogger(__name__) +class TaskParams(BlueapiBaseModel): + args: tuple[Any, ...] = () + kwargs: Mapping[str, Any] = Field(default_factory=dict) + + class Task(BlueapiBaseModel): """ Task that will run a plan """ name: str = Field(description="Name of plan to run") - params: Mapping[str, Any] = Field( - description="Values for parameters to plan, if any", default_factory=dict + params: TaskParams = Field( + description="Values for parameters to plan, if any", default_factory=TaskParams ) metadata: dict[str, Any] = Field( description="Any metadata to apply to all runs within this task", default_factory=dict, ) - def prepare_params(self, ctx: BlueskyContext) -> Mapping[str, Any]: - model = _lookup_params(ctx, self) - # Re-create dict manually to avoid nesting in model_dump output - return {field: getattr(model, field) for field in model.__pydantic_fields__} + def prepare_params( + self, ctx: BlueskyContext + ) -> tuple[tuple[Any, ...], Mapping[str, Any]]: + return _lookup_params(ctx, self) def do_task(self, ctx: BlueskyContext) -> None: LOGGER.info( f"Asked to run plan {self.name} with {self.params} and " f"metadata {self.metadata} for all runs" ) - - func = ctx.plan_functions[self.name] - prepared_params = self.prepare_params(ctx) + plan = ctx.plan_functions[self.name] + prepared_args, prepared_kwargs = self.prepare_params(ctx) ctx.run_engine.md.update(self.metadata) - result = ctx.run_engine(func(**prepared_params)) + result = ctx.run_engine(plan(*prepared_args, **prepared_kwargs)) if isinstance(result, tuple): # pragma: no cover # this is never true if the run_engine is configured correctly return None return result.plan_result -def _lookup_params(ctx: BlueskyContext, task: Task) -> BaseModel: +def _lookup_params( + ctx: BlueskyContext, task: Task +) -> tuple[tuple[Any, ...], Mapping[str, Any]]: + """ + Validate and prepare the arguments for a plan. """ - Checks plan parameters against context + plan = ctx.plans[task.name] + func = ctx.plan_functions[task.name] - Args: - ctx: Context holding plans and devices - plan: Plan object including schema - params: Parameter values to be validated against schema + sig = signature(func) + bound = sig.bind(*task.params.args, **task.params.kwargs) - Returns: - Mapping[str, Any]: _description_ - """ + # Only validate explicitly supplied arguments. This allows Pydantic's + # default/default_factory to provide injected defaults. + adapter = TypeAdapter(plan.model) + validated = adapter.validate_python(bound.arguments) - plan = ctx.plans[task.name] - model = plan.model - adapter = TypeAdapter(model) - return adapter.validate_python(task.params) + args: list[Any] = [] + kwargs: dict[str, Any] = {} + + for name, parameter in sig.parameters.items(): + supplied = name in bound.arguments + + if supplied: + value = getattr(validated, name) + + match parameter.kind: + case Parameter.POSITIONAL_ONLY: + args.append(value) + + case Parameter.POSITIONAL_OR_KEYWORD: + if name in task.params.kwargs: + kwargs[name] = value + else: + args.append(value) + + case Parameter.VAR_POSITIONAL: + args.extend(value) + + case Parameter.KEYWORD_ONLY: + kwargs[name] = value + + case Parameter.VAR_KEYWORD: + kwargs.update(value) + + else: + # Let the generated Pydantic model provide the default. + value = getattr(validated, name) + kwargs[name] = value + + return tuple(args), kwargs diff --git a/tests/unit_tests/service/test_rest_api.py b/tests/unit_tests/service/test_rest_api.py index a10d5ab3cf..9b6bedf1ba 100644 --- a/tests/unit_tests/service/test_rest_api.py +++ b/tests/unit_tests/service/test_rest_api.py @@ -47,7 +47,7 @@ ) from blueapi.service.runner import WorkerDispatcher from blueapi.worker.event import TaskStatus, WorkerEvent, WorkerState -from blueapi.worker.task import Task +from blueapi.worker.task import Task, TaskParams from blueapi.worker.task_worker import TrackableTask @@ -395,7 +395,10 @@ def test_put_plan_fails_if_not_idle(mock_runner: Mock, client: TestClient) -> No def test_get_tasks(mock_runner: Mock, client: TestClient) -> None: tasks = [ - TrackableTask(task_id="0", task=Task(name="sleep", params={"time": 0.0})), + TrackableTask( + task_id="0", + task=Task(name="sleep", params=TaskParams(kwargs={"time": 0.0})), + ), TrackableTask( task_id="1", task=Task(name="first_task"), @@ -418,7 +421,7 @@ def test_get_tasks(mock_runner: Mock, client: TestClient) -> None: "request_id": None, "task": { "name": "sleep", - "params": {"time": 0.0}, + "params": {"args": [], "kwargs": {"time": 0.0}}, "metadata": {}, }, "outcome": None, @@ -431,7 +434,7 @@ def test_get_tasks(mock_runner: Mock, client: TestClient) -> None: "request_id": None, "task": { "name": "first_task", - "params": {}, + "params": {"args": [], "kwargs": {}}, "metadata": {}, }, "outcome": None, @@ -463,7 +466,7 @@ def test_get_tasks_by_status(mock_runner: Mock, client: TestClient) -> None: "request_id": None, "task": { "name": "third_task", - "params": {}, + "params": {"args": [], "kwargs": {}}, "metadata": {}, }, "outcome": None, @@ -627,7 +630,7 @@ def test_get_task(mock_runner: Mock, client: TestClient): "request_id": None, "task": { "name": "third_task", - "params": {}, + "params": {"args": [], "kwargs": {}}, "metadata": { "foo": "bar", }, @@ -674,7 +677,7 @@ def test_get_all_tasks(mock_runner: Mock, client: TestClient): "task_id": task_id, "task": { "name": "third_task", - "params": {}, + "params": {"args": [], "kwargs": {}}, "metadata": {}, }, "is_complete": False, diff --git a/tests/unit_tests/worker/test_task_worker.py b/tests/unit_tests/worker/test_task_worker.py index 5e6553d2fe..fe1383ed86 100644 --- a/tests/unit_tests/worker/test_task_worker.py +++ b/tests/unit_tests/worker/test_task_worker.py @@ -27,6 +27,7 @@ from blueapi.utils.base_model import BlueapiBaseModel from blueapi.worker import ( Task, + TaskParams, TaskStatus, TaskWorker, TrackableTask, @@ -37,16 +38,16 @@ ) from blueapi.worker.event import TaskResult, TaskStatusEnum -_SIMPLE_TASK = Task(name="sleep", params={"time": 0.0}) -_LONG_TASK = Task(name="sleep", params={"time": 1.0}) +_SIMPLE_TASK = Task(name="sleep", params=TaskParams(kwargs={"time": 0.0})) +_LONG_TASK = Task(name="sleep", params=TaskParams(kwargs={"time": 1.0})) _INDEFINITE_TASK = Task( name="set_absolute", - params={"movable": "fake_device", "value": 4.0}, + params=TaskParams(kwargs={"movable": "fake_device", "value": 4.0}), ) -_FAILING_TASK = Task(name="failing_plan", params={}) +_FAILING_TASK = Task(name="failing_plan", params=TaskParams()) _TASK_WITH_METADATA = Task( name="sleep", - params={"time": 0.0}, + params=TaskParams(kwargs={"time": 0.0}), metadata={ "foo": "bar", "baz": 0, @@ -522,7 +523,9 @@ def assert_running_count_plan_produces_ordered_worker_and_data_events( task: Task | None = None, timeout: float = 5.0, ) -> None: - default_task = Task(name="count", params={"detectors": ["motor"], "num": 1}) + default_task = Task( + name="count", params=TaskParams(kwargs={"detectors": ["motor"], "num": 1}) + ) task = task or default_task event_streams: list[EventStream[Any, int]] = [ @@ -633,7 +636,8 @@ def test_get_tasks(worker: TaskWorker, status, expected_task_ids): "task1": TrackableTask( task_id="task1", task=Task( - name="set_absolute", params={"movable": "fake_device", "value": 4.0} + name="set_absolute", + params=TaskParams(kwargs={"movable": "fake_device", "value": 4.0}), ), is_complete=False, is_pending=False, @@ -641,7 +645,8 @@ def test_get_tasks(worker: TaskWorker, status, expected_task_ids): "task2": TrackableTask( task_id="task2", task=Task( - name="set_absolute", params={"movable": "fake_device", "value": 4.0} + name="set_absolute", + params=TaskParams(kwargs={"movable": "fake_device", "value": 4.0}), ), is_complete=False, is_pending=True, @@ -651,7 +656,8 @@ def test_get_tasks(worker: TaskWorker, status, expected_task_ids): "task3": TrackableTask( task_id="task3", task=Task( - name="set_absolute", params={"movable": "fake_device", "value": 4.0} + name="set_absolute", + params=TaskParams(kwargs={"movable": "fake_device", "value": 4.0}), ), is_complete=True, is_pending=False, @@ -737,8 +743,8 @@ def injected_device_plan( yield from () context.register_plan(injected_device_plan) - params = Task(name="injected_device_plan").prepare_params(context) - assert params["dev"] == fake_device + args, kwargs = Task(name="injected_device_plan").prepare_params(context) + assert kwargs["dev"] == fake_device def test_injected_devices_plan_model( @@ -853,9 +859,9 @@ def test_injected_composite_devices_are_found( context: BlueskyContext, ): context.register_plan(injected_device_plan) - params = Task(name="injected_device_plan").prepare_params(context) - assert params["composite"].fake_device == fake_device - assert params["composite"].second_fake_device == second_fake_device + args, kwargs = Task(name="injected_device_plan").prepare_params(context) + assert kwargs["composite"].fake_device == fake_device + assert kwargs["composite"].second_fake_device == second_fake_device def test_injected_composite_devices_plan_model( @@ -874,9 +880,9 @@ def test_injected_composite_with_pydantic_dataclass( second_fake_device: FakeDevice, ): context.register_plan(injected_dataclass_device_plan) - params = Task(name="injected_dataclass_device_plan").prepare_params(context) - assert params["composite"].fake_device == fake_device - assert params["composite"].second_fake_device == second_fake_device + args, kwargs = Task(name="injected_dataclass_device_plan").prepare_params(context) + assert kwargs["composite"].fake_device == fake_device + assert kwargs["composite"].second_fake_device == second_fake_device def test_injected_composite_with_standard_dataclass( @@ -885,11 +891,11 @@ def test_injected_composite_with_standard_dataclass( second_fake_device: FakeDevice, ): context.register_plan(injected_standard_dataclass_device_plan) - params = Task(name="injected_standard_dataclass_device_plan").prepare_params( + args, kwargs = Task(name="injected_standard_dataclass_device_plan").prepare_params( context ) - assert params["composite"].fake_device == fake_device - assert params["composite"].second_fake_device == second_fake_device + assert kwargs["composite"].fake_device == fake_device + assert kwargs["composite"].second_fake_device == second_fake_device def test_plan_module_with_composite_devices_can_be_loaded_before_device_module( @@ -900,9 +906,11 @@ def test_plan_module_with_composite_devices_can_be_loaded_before_device_module( context_without_devices.register_plan(injected_device_plan) context_without_devices.register_device(fake_device) context_without_devices.register_device(second_fake_device) - params = Task(name="injected_device_plan").prepare_params(context_without_devices) - assert params["composite"].fake_device == fake_device - assert params["composite"].second_fake_device == second_fake_device + args, kwargs = Task(name="injected_device_plan").prepare_params( + context_without_devices + ) + assert kwargs["composite"].fake_device == fake_device + assert kwargs["composite"].second_fake_device == second_fake_device @pytest.mark.parametrize( From fc09b961448b6a80eabb7cbc5a91be67e1048a59 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 9 Sep 2026 15:20:32 +0000 Subject: [PATCH 02/31] Add tests to cover *args case --- src/blueapi/worker/task.py | 10 +-- tests/unit_tests/worker/test_task_worker.py | 70 +++++++++++++++++++++ 2 files changed, 75 insertions(+), 5 deletions(-) diff --git a/src/blueapi/worker/task.py b/src/blueapi/worker/task.py index 648f442947..5c39508194 100644 --- a/src/blueapi/worker/task.py +++ b/src/blueapi/worker/task.py @@ -1,5 +1,5 @@ import logging -from collections.abc import Mapping +from collections.abc import Mapping, Sequence from inspect import Parameter, signature from typing import Any @@ -12,7 +12,7 @@ class TaskParams(BlueapiBaseModel): - args: tuple[Any, ...] = () + args: Sequence[Any] = [] kwargs: Mapping[str, Any] = Field(default_factory=dict) @@ -32,7 +32,7 @@ class Task(BlueapiBaseModel): def prepare_params( self, ctx: BlueskyContext - ) -> tuple[tuple[Any, ...], Mapping[str, Any]]: + ) -> tuple[list[Any], Mapping[str, Any]]: return _lookup_params(ctx, self) def do_task(self, ctx: BlueskyContext) -> None: @@ -52,7 +52,7 @@ def do_task(self, ctx: BlueskyContext) -> None: def _lookup_params( ctx: BlueskyContext, task: Task -) -> tuple[tuple[Any, ...], Mapping[str, Any]]: +) -> tuple[list[Any], Mapping[str, Any]]: """ Validate and prepare the arguments for a plan. """ @@ -100,4 +100,4 @@ def _lookup_params( value = getattr(validated, name) kwargs[name] = value - return tuple(args), kwargs + return args, kwargs diff --git a/tests/unit_tests/worker/test_task_worker.py b/tests/unit_tests/worker/test_task_worker.py index fe1383ed86..6df6a3b839 100644 --- a/tests/unit_tests/worker/test_task_worker.py +++ b/tests/unit_tests/worker/test_task_worker.py @@ -970,3 +970,73 @@ def test_worker_event_task_id(): def test_worker_event_no_task_id(): event = WorkerEvent(state=WorkerState.IDLE, task_status=None) assert event.task_id is None + + +def test_task_worker_passes_positional_args( + context: BlueskyContext, +) -> None: + + def positional_plan(value: int) -> MsgGenerator: + yield from () + + context.register_plan(positional_plan) + + task = Task(name="positional_plan", params=TaskParams(args=[42])) + args, kwargs = task.prepare_params(context) + + assert args == [42] + assert kwargs == {} + + +def test_task_worker_passes_multiple_positional_args( + context: BlueskyContext, +) -> None: + + def positional_plan(first: int, second: int) -> MsgGenerator: + yield from () + + context.register_plan(positional_plan) + task = Task(name="positional_plan", params=TaskParams(args=[1, 2])) + args, kwargs = task.prepare_params(context) + + assert args == [1, 2] + assert kwargs == {} + + +def test_task_worker_passes_positional_and_keyword_args( + context: BlueskyContext, +) -> None: + + def mixed_plan( + first: int, + second: int, + *, + third: int, + ) -> MsgGenerator: + yield from () + + context.register_plan(mixed_plan) + task = Task( + name="mixed_plan", + params=TaskParams(args=[1, 2], kwargs={"third": 3}), + ) + args, kwargs = task.prepare_params(context) + assert args == [1, 2, 3] + assert kwargs == {} + + +def test_task_worker_resolves_positional_device( + context: BlueskyContext, + fake_device: FakeDevice, +) -> None: + def positional_device_plan(device: FakeDevice) -> MsgGenerator: + yield from () + + context.register_plan(positional_device_plan) + + task = Task( + name="positional_device_plan", params=TaskParams(args=[fake_device.name]) + ) + args, kwargs = task.prepare_params(context) + assert args == [fake_device] + assert kwargs == {} From d6388843b6573839626ace0bc8d9b64fbb90269d Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 9 Sep 2026 15:27:15 +0000 Subject: [PATCH 03/31] Simplify logic --- src/blueapi/worker/task.py | 108 +++++++++++++++++++------------------ 1 file changed, 55 insertions(+), 53 deletions(-) diff --git a/src/blueapi/worker/task.py b/src/blueapi/worker/task.py index 5c39508194..7e9662cd30 100644 --- a/src/blueapi/worker/task.py +++ b/src/blueapi/worker/task.py @@ -33,71 +33,73 @@ class Task(BlueapiBaseModel): def prepare_params( self, ctx: BlueskyContext ) -> tuple[list[Any], Mapping[str, Any]]: - return _lookup_params(ctx, self) + """ + Checks plan parameters against context - def do_task(self, ctx: BlueskyContext) -> None: - LOGGER.info( - f"Asked to run plan {self.name} with {self.params} and " - f"metadata {self.metadata} for all runs" - ) - plan = ctx.plan_functions[self.name] - prepared_args, prepared_kwargs = self.prepare_params(ctx) - ctx.run_engine.md.update(self.metadata) - result = ctx.run_engine(plan(*prepared_args, **prepared_kwargs)) - if isinstance(result, tuple): # pragma: no cover - # this is never true if the run_engine is configured correctly - return None - return result.plan_result + Args: + ctx: Context holding plans and devices + plan: Plan object including schema + params: Parameter values to be validated against schema + Returns: + Mapping[str, Any]: _description_ + """ + plan = ctx.plans[self.name] + func = ctx.plan_functions[self.name] -def _lookup_params( - ctx: BlueskyContext, task: Task -) -> tuple[list[Any], Mapping[str, Any]]: - """ - Validate and prepare the arguments for a plan. - """ - plan = ctx.plans[task.name] - func = ctx.plan_functions[task.name] + sig = signature(func) + bound = sig.bind(*self.params.args, **self.params.kwargs) + + # Only validate explicitly supplied arguments. This allows Pydantic's + # default/default_factory to provide injected defaults. + adapter = TypeAdapter(plan.model) + validated = adapter.validate_python(bound.arguments) - sig = signature(func) - bound = sig.bind(*task.params.args, **task.params.kwargs) + args: list[Any] = [] + kwargs: dict[str, Any] = {} - # Only validate explicitly supplied arguments. This allows Pydantic's - # default/default_factory to provide injected defaults. - adapter = TypeAdapter(plan.model) - validated = adapter.validate_python(bound.arguments) + for name, parameter in sig.parameters.items(): + supplied = name in bound.arguments - args: list[Any] = [] - kwargs: dict[str, Any] = {} + if supplied: + value = getattr(validated, name) - for name, parameter in sig.parameters.items(): - supplied = name in bound.arguments + match parameter.kind: + case Parameter.POSITIONAL_ONLY: + args.append(value) - if supplied: - value = getattr(validated, name) + case Parameter.POSITIONAL_OR_KEYWORD: + if name in self.params.kwargs: + kwargs[name] = value + else: + args.append(value) - match parameter.kind: - case Parameter.POSITIONAL_ONLY: - args.append(value) + case Parameter.VAR_POSITIONAL: + args.extend(value) - case Parameter.POSITIONAL_OR_KEYWORD: - if name in task.params.kwargs: + case Parameter.KEYWORD_ONLY: kwargs[name] = value - else: - args.append(value) - case Parameter.VAR_POSITIONAL: - args.extend(value) + case Parameter.VAR_KEYWORD: + kwargs.update(value) - case Parameter.KEYWORD_ONLY: - kwargs[name] = value + else: + # Let the generated Pydantic model provide the default. + value = getattr(validated, name) + kwargs[name] = value - case Parameter.VAR_KEYWORD: - kwargs.update(value) + return args, kwargs - else: - # Let the generated Pydantic model provide the default. - value = getattr(validated, name) - kwargs[name] = value - - return args, kwargs + def do_task(self, ctx: BlueskyContext) -> None: + LOGGER.info( + f"Asked to run plan {self.name} with {self.params} and " + f"metadata {self.metadata} for all runs" + ) + plan = ctx.plan_functions[self.name] + prepared_args, prepared_kwargs = self.prepare_params(ctx) + ctx.run_engine.md.update(self.metadata) + result = ctx.run_engine(plan(*prepared_args, **prepared_kwargs)) + if isinstance(result, tuple): # pragma: no cover + # this is never true if the run_engine is configured correctly + return None + return result.plan_result From d3d45e107f42274860b2a7e73e0957bd7bfc5ed3 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 9 Sep 2026 15:27:19 +0000 Subject: [PATCH 04/31] Fix test --- tests/unit_tests/worker/test_task_worker.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/unit_tests/worker/test_task_worker.py b/tests/unit_tests/worker/test_task_worker.py index 6df6a3b839..b4f2460af5 100644 --- a/tests/unit_tests/worker/test_task_worker.py +++ b/tests/unit_tests/worker/test_task_worker.py @@ -1021,8 +1021,8 @@ def mixed_plan( params=TaskParams(args=[1, 2], kwargs={"third": 3}), ) args, kwargs = task.prepare_params(context) - assert args == [1, 2, 3] - assert kwargs == {} + assert args == [1, 2] + assert kwargs == {"third": 3} def test_task_worker_resolves_positional_device( From 44ce2fa5ccefa6feb5074d76c5c4ef424ca58c17 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 9 Sep 2026 15:29:31 +0000 Subject: [PATCH 05/31] Use default_factory for args --- src/blueapi/worker/task.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/blueapi/worker/task.py b/src/blueapi/worker/task.py index 7e9662cd30..4c64698930 100644 --- a/src/blueapi/worker/task.py +++ b/src/blueapi/worker/task.py @@ -12,7 +12,7 @@ class TaskParams(BlueapiBaseModel): - args: Sequence[Any] = [] + args: Sequence[Any] = Field(default_factory=lambda: []) kwargs: Mapping[str, Any] = Field(default_factory=dict) From 32cc60e2e9274952bef8d8717b1a3af79eaca507 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 9 Sep 2026 16:24:29 +0000 Subject: [PATCH 06/31] Propergate change --- src/blueapi/cli/cli.py | 6 +- src/blueapi/client/client.py | 32 +++++------ src/blueapi/service/main.py | 4 +- src/blueapi/service/model.py | 8 +-- tests/system_tests/test_blueapi_system.py | 69 ++++++++++++----------- tests/unit_tests/cli/test_cli.py | 11 ++-- tests/unit_tests/client/test_client.py | 46 ++++++++++++--- tests/unit_tests/client/test_rest.py | 12 ++-- tests/unit_tests/service/test_rest_api.py | 12 ++-- 9 files changed, 123 insertions(+), 77 deletions(-) diff --git a/src/blueapi/cli/cli.py b/src/blueapi/cli/cli.py index c3f9bde13f..a3a9e56821 100644 --- a/src/blueapi/cli/cli.py +++ b/src/blueapi/cli/cli.py @@ -39,7 +39,7 @@ from blueapi.log import set_up_logging from blueapi.service.authentication import SessionCacheManager, SessionManager from blueapi.service.model import DeviceResponse, PlanResponse, SourceInfo, TaskRequest -from blueapi.worker import ProgressEvent, WorkerEvent +from blueapi.worker import ProgressEvent, TaskParams, WorkerEvent from blueapi.worker.event import TaskError, TaskResult from .scratch import setup_scratch @@ -374,7 +374,9 @@ def run_plan( client = cast(BlueapiClient, obj["client"]) task = TaskRequest( - name=name, params=parameters, instrument_session=instrument_session + name=name, + params=TaskParams(kwargs=parameters), + instrument_session=instrument_session, ) try: diff --git a/src/blueapi/client/client.py b/src/blueapi/client/client.py index 66a79d8dd6..3d77a93daa 100644 --- a/src/blueapi/client/client.py +++ b/src/blueapi/client/client.py @@ -31,6 +31,7 @@ PlanResponse, PythonEnvironmentResponse, SourceInfo, + TaskParams, TaskRequest, TaskResponse, TasksListResponse, @@ -180,7 +181,7 @@ def properties(self) -> dict[str, Any]: def required(self) -> list[str]: return self.model.parameter_schema.get("required", []) - def _build_args(self, *args, **kwargs): + def _build_args(self, *args, **kwargs) -> TaskParams: log.info( "Building args for %s, using %s and %s", "[" + ",".join(self.properties) + "]", @@ -188,27 +189,26 @@ def _build_args(self, *args, **kwargs): kwargs, ) - if len(args) > len(self.properties): + properties = list(self.properties) + + if len(args) > len(properties): raise TypeError(f"{self.name} got too many arguments") - if extra := {k for k in kwargs if k not in self.properties}: + + if extra := {k for k in kwargs if k not in properties}: raise TypeError(f"{self.name} got unexpected arguments: {extra}") - params = {} - # Initially fill parameters using positional args assuming the order - # from the parameter_schema - for req, arg in zip(self.properties, args, strict=False): - params[req] = arg + positional_names = properties[: len(args)] - # Then append any values given via kwargs - for key, value in kwargs.items(): - # If we've already assumed a positional arg was this value, bail out - if key in params: - raise TypeError(f"{self.name} got multiple values for {key}") - params[key] = value + if duplicate := set(positional_names) & kwargs.keys(): + name = next(iter(duplicate)) + raise TypeError(f"{self.name} got multiple values for {name}") - if missing := {k for k in self.required if k not in params}: + supplied = set(positional_names) | kwargs.keys() + + if missing := set(self.required) - supplied: raise TypeError(f"Missing argument(s) for {missing}") - return params + + return TaskParams(args=args, kwargs=kwargs) def __repr__(self) -> str: required = set(self.required) diff --git a/src/blueapi/service/main.py b/src/blueapi/service/main.py index 0d8fbf4774..81b7fd1cc7 100644 --- a/src/blueapi/service/main.py +++ b/src/blueapi/service/main.py @@ -55,7 +55,7 @@ Unauthorized, Update, ) -from blueapi.worker import TrackableTask, WorkerState +from blueapi.worker import TaskParams, TrackableTask, WorkerState from blueapi.worker.event import ProgressEvent, TaskStatusEnum, WorkerEvent from blueapi.worker.worker_errors import WorkerBusyError @@ -320,7 +320,7 @@ def get_device_by_name( example_task_request = TaskRequest( name="count", - params={"detectors": ["x"]}, + params=TaskParams(kwargs={"detectors": ["x"]}), instrument_session="cm12345-1", ) diff --git a/src/blueapi/service/model.py b/src/blueapi/service/model.py index dbaa1d9651..7d7aaabc1d 100644 --- a/src/blueapi/service/model.py +++ b/src/blueapi/service/model.py @@ -1,5 +1,5 @@ import uuid -from collections.abc import Iterable, Mapping +from collections.abc import Iterable from enum import StrEnum from typing import Annotated, Any @@ -11,7 +11,7 @@ from blueapi.core import BLUESKY_PROTOCOLS, Device, Plan from blueapi.core.context import generic_bounds from blueapi.utils import BlueapiBaseModel -from blueapi.worker import WorkerState +from blueapi.worker import TaskParams, WorkerState from blueapi.worker.task_worker import TaskWorker, TrackableTask _UNKNOWN_NAME = "UNKNOWN" @@ -64,8 +64,8 @@ class TaskRequest(BlueapiBaseModel): """ name: str = Field(description="Name of plan to run") - params: Mapping[str, Any] = Field( - description="Values for parameters to plan, if any", default_factory=dict + params: TaskParams = Field( + description="Values for parameters to plan, if any", default_factory=TaskParams ) instrument_session: str = Field( description="Instrument session associated with this task", diff --git a/tests/system_tests/test_blueapi_system.py b/tests/system_tests/test_blueapi_system.py index 8c4dc19cf9..55237850e7 100644 --- a/tests/system_tests/test_blueapi_system.py +++ b/tests/system_tests/test_blueapi_system.py @@ -32,6 +32,7 @@ from blueapi.service.model import ( DeviceResponse, PlanResponse, + TaskParams, TaskRequest, TaskResponse, WorkerTask, @@ -122,7 +123,7 @@ def task_factory( ) -> TaskRequest: return TaskRequest( name="sleep", - params={"time": time}, + params=TaskParams(kwargs={"time": time}), instrument_session=instrument_session if instrument_session else VALID_INSTRUMENT_SESSION[user], @@ -365,7 +366,7 @@ def test_create_task_validation_error(rest_client: BlueapiRestClient): rest_client.create_task( TaskRequest( name="Not-exists", - params={"Not-exists": 0.0}, + params=TaskParams(kwargs={"Not-exists": 0.0}), instrument_session="Not-exists", ) ) @@ -542,7 +543,7 @@ def test_delete_current_environment(client: BlueapiClient): ( TaskRequest( name="count", - params={"detectors": ["det"], "num": 3}, + params=TaskParams(kwargs={"detectors": ["det"], "num": 3}), instrument_session=VALID_INSTRUMENT_SESSION[User.alice], ), User.alice, @@ -550,27 +551,29 @@ def test_delete_current_environment(client: BlueapiClient): ( TaskRequest( name="spec_scan", - params={ - "detectors": ["det"], - "spec": { - "outer": { - "axis": "stage.x", - "start": 0.0, - "stop": 0.4, - "num": 2, - "type": "Linspace", - }, - "inner": { - "axis": "stage.theta", - "start": 5.0, - "stop": 5.3, - "num": 3, - "type": "Linspace", + params=TaskParams( + kwargs={ + "detectors": ["det"], + "spec": { + "outer": { + "axis": "stage.x", + "start": 0.0, + "stop": 0.4, + "num": 2, + "type": "Linspace", + }, + "inner": { + "axis": "stage.theta", + "start": 5.0, + "stop": 5.3, + "num": 3, + "type": "Linspace", + }, + "gap": True, + "type": "Product", }, - "gap": True, - "type": "Product", - }, - }, + } + ), instrument_session=VALID_INSTRUMENT_SESSION[User.bob], ), User.bob, @@ -638,10 +641,12 @@ def on_event(event: AnyEvent) -> None: [ TaskRequest( name="set_absolute", - params={ - "movable": "stage.x", - "value": 1.0, - }, + params=TaskParams( + kwargs={ + "movable": "stage.x", + "value": 1.0, + } + ), instrument_session=VALID_INSTRUMENT_SESSION[User.alice], ), ], @@ -663,11 +668,11 @@ def test_task_submission_after_invalid_task(client_with_stomp: BlueapiClient): res = client_with_stomp.run_task( TaskRequest( name="count", - params={ - "detectors": [ - "det", - ], - }, + params=TaskParams( + kwargs={ + "detectors": ["det"], + } + ), instrument_session=VALID_INSTRUMENT_SESSION[User.alice], ) ) diff --git a/tests/unit_tests/cli/test_cli.py b/tests/unit_tests/cli/test_cli.py index 28cb4fbc92..7131bae224 100644 --- a/tests/unit_tests/cli/test_cli.py +++ b/tests/unit_tests/cli/test_cli.py @@ -50,6 +50,7 @@ TaskRequest, TaskResponse, ) +from blueapi.worker import TaskParams from blueapi.worker.event import ( ProgressEvent, TaskError, @@ -210,7 +211,7 @@ def test_invalid_config_via_env(runner: CliRunner): def test_submit_plan(runner: CliRunner): body_data = { "name": "sleep", - "params": {"time": 5}, + "params": {"args": [], "kwargs": {"time": 5}}, "instrument_session": "cm12345-1", } @@ -270,7 +271,7 @@ def test_run_plan(stomp_client: StompClient, runner: CliRunner): matchers.json_params_matcher( { "name": "sleep", - "params": {"time": 3}, + "params": {"args": [], "kwargs": {"time": 3}}, "instrument_session": "cm12345-1", } ) @@ -386,7 +387,7 @@ def test_run_plan_feedback( ) bc.add_callback.assert_called_once() bc.run_task.assert_called_once_with( - TaskRequest(name="name", params={}, instrument_session="cm12345-1"), + TaskRequest(name="name", params=TaskParams(), instrument_session="cm12345-1"), ) assert res.exit_code == 0 assert res.stdout == message @@ -400,7 +401,7 @@ def test_run_plan_background_without_stomp(runner: CliRunner): matchers.json_params_matcher( { "name": "sleep", - "params": {"time": 3}, + "params": {"args": [], "kwargs": {"time": 3}}, "instrument_session": "cm12345-1", } ) @@ -503,7 +504,7 @@ def test_can_pass_an_instrument_session_with_an_environment_variable( mock_create_task.assert_called_once_with( TaskRequest( name="sleep", - params={"time": 5.0}, + params=TaskParams(kwargs={"time": 5.0}), instrument_session="cm12345-1", ) ) diff --git a/tests/unit_tests/client/test_client.py b/tests/unit_tests/client/test_client.py index d5b0493adc..13502270cc 100644 --- a/tests/unit_tests/client/test_client.py +++ b/tests/unit_tests/client/test_client.py @@ -42,7 +42,14 @@ TasksListResponse, WorkerTask, ) -from blueapi.worker import ProgressEvent, Task, TrackableTask, WorkerEvent, WorkerState +from blueapi.worker import ( + ProgressEvent, + Task, + TaskParams, + TrackableTask, + WorkerEvent, + WorkerState, +) from blueapi.worker.event import TaskError, TaskResult, TaskStatus PLANS = PlanResponse( @@ -72,7 +79,7 @@ ] ) DEVICE = DeviceModel(name="foo", protocols=[]) -TASK = TrackableTask(task_id="foo", task=Task(name="bar", params={})) +TASK = TrackableTask(task_id="foo", task=Task(name="bar", params=TaskParams())) TASKS = TasksListResponse(tasks=[TASK]) ACTIVE_TASK = WorkerTask(task_id="bar") ENVIRONMENT_ID = uuid.uuid4() @@ -900,11 +907,36 @@ def test_plan_empty_fallback_help_text(client): @pytest.mark.parametrize( "args,kwargs,params", [ - p((1,), {}, {"one": 1}, id="required_as_positional"), - p((), {"one": 7}, {"one": 7}, id="required_as_keyword"), - p((1,), {"two": 23}, {"one": 1, "two": 23}, id="all_as_mixed_args_kwargs"), - p((1, 2), {}, {"one": 1, "two": 2}, id="all_as_positional"), - p((), {"one": 21, "two": 42}, {"one": 21, "two": 42}, id="all_as_keyword"), + p( + (1,), + {}, + TaskParams(args=(1,), kwargs={}), + id="required_as_positional", + ), + p( + (), + {"one": 7}, + TaskParams(args=(), kwargs={"one": 7}), + id="required_as_keyword", + ), + p( + (1,), + {"two": 23}, + TaskParams(args=(1,), kwargs={"two": 23}), + id="all_as_mixed_args_kwargs", + ), + p( + (1, 2), + {}, + TaskParams(args=(1, 2), kwargs={}), + id="all_as_positional", + ), + p( + (), + {"one": 21, "two": 42}, + TaskParams(args=(), kwargs={"one": 21, "two": 42}), + id="all_as_keyword", + ), ], ) def test_plan_param_mapping(args, kwargs, params): diff --git a/tests/unit_tests/client/test_rest.py b/tests/unit_tests/client/test_rest.py index 6ecfbaa765..7aa2870886 100644 --- a/tests/unit_tests/client/test_rest.py +++ b/tests/unit_tests/client/test_rest.py @@ -40,11 +40,11 @@ WorkerTask, ) from blueapi.worker.event import WorkerState -from blueapi.worker.task import Task +from blueapi.worker.task import Task, TaskParams from blueapi.worker.task_worker import TrackableTask TASK_REQUEST = TaskRequest( - name="foo", params={"one": "two"}, instrument_session="cm12345-1" + name="foo", params=TaskParams(kwargs={"one": "two"}), instrument_session="cm12345-1" ) @@ -94,7 +94,9 @@ def test_create_task_serialization(): request = TaskRequest( name="demo", instrument_session="cm12345-1", - params={"devices": [DeviceRef(name="foo", cache=Mock(), model=Mock())]}, + params=TaskParams( + kwargs={"devices": [DeviceRef(name="foo", cache=Mock(), model=Mock())]} + ), ) BlueapiRestClient.create_task(rest, request) @@ -107,7 +109,7 @@ def test_create_task_serialization(): data={ "name": "demo", "instrument_session": "cm12345-1", - "params": {"devices": ["foo"]}, + "params": {"args": [], "kwargs": {"devices": ["foo"]}}, }, ) @@ -120,7 +122,7 @@ class CustomType: request = TaskRequest( name="demo", instrument_session="cm12345-1", - params={"devices": [CustomType()]}, + params=TaskParams(kwargs={"devices": [CustomType()]}), ) with pytest.raises(PydanticSerializationError, match="not serializable"): diff --git a/tests/unit_tests/service/test_rest_api.py b/tests/unit_tests/service/test_rest_api.py index 9b6bedf1ba..712b246f5c 100644 --- a/tests/unit_tests/service/test_rest_api.py +++ b/tests/unit_tests/service/test_rest_api.py @@ -172,7 +172,7 @@ def test_rest_config_with_cors( ): task = TaskRequest( name="my-plan", - params={"id": "x"}, + params=TaskParams(kwargs={"id": "x"}), instrument_session=FAKE_INSTRUMENT_SESSION, ) task_id = "f8424be3-203c-494e-b22f-219933b4fa67" @@ -287,7 +287,7 @@ def test_get_non_existent_device_by_name(mock_runner: Mock, client: TestClient) def test_create_task(mock_runner: Mock, client: TestClient) -> None: task = TaskRequest( name="count", - params={"detectors": ["x"]}, + params=TaskParams(kwargs={"detectors": ["x"]}), instrument_session=FAKE_INSTRUMENT_SESSION, ) task_id = str(uuid.uuid4()) @@ -306,7 +306,11 @@ def test_submit_task_requires_permission( mock_opa_client: Mock, access_token: str, ): - task = TaskRequest(name="sleep", params={"time": 2}, instrument_session="cm12345-2") + task = TaskRequest( + name="sleep", + params=TaskParams(kwargs={"time": 2}), + instrument_session="cm12345-2", + ) client_with_opa.headers["Authorization"] = f"Bearer {access_token}" mock_opa_client.can_submit_task.side_effect = HTTPException(status_code=403) mock_runner.run.side_effect = RuntimeError("Task should not be submitted") @@ -323,7 +327,7 @@ def test_create_task_inserts_auth_metadata( ) -> None: task = TaskRequest( name="count", - params={"detectors": ["x"]}, + params=TaskParams(kwargs={"detectors": ["x"]}), instrument_session=FAKE_INSTRUMENT_SESSION, ) client_with_auth.follow_redirects = False From 42bad034038810251b17ca6c8838bfd25ed902bd Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 9 Sep 2026 16:30:03 +0000 Subject: [PATCH 07/31] Fix lint --- tests/unit_tests/cli/test_cli.py | 2 +- tests/unit_tests/service/test_authorization.py | 3 ++- tests/unit_tests/service/test_interface.py | 6 +++--- tests/unit_tests/service/test_protocol.py | 5 ++++- 4 files changed, 10 insertions(+), 6 deletions(-) diff --git a/tests/unit_tests/cli/test_cli.py b/tests/unit_tests/cli/test_cli.py index 7131bae224..d491d45d7d 100644 --- a/tests/unit_tests/cli/test_cli.py +++ b/tests/unit_tests/cli/test_cli.py @@ -1490,6 +1490,6 @@ def test_run_ws_runs_blocking_plan(mock_client: Mock, runner: CliRunner): bc.add_callback.assert_called_once() bc.run_task.assert_not_called() bc.run_blocking.assert_called_once_with( - TaskRequest(name="name", params={}, instrument_session="cm12345-1"), + TaskRequest(name="name", params=TaskParams(), instrument_session="cm12345-1"), ) assert res.exit_code == 0 diff --git a/tests/unit_tests/service/test_authorization.py b/tests/unit_tests/service/test_authorization.py index a2e602f211..2be684c6bd 100644 --- a/tests/unit_tests/service/test_authorization.py +++ b/tests/unit_tests/service/test_authorization.py @@ -14,6 +14,7 @@ validate_tiled_config, ) from blueapi.service.model import TaskRequest +from blueapi.worker import TaskParams # Reusable client patch decorator patch_client_session = patch( @@ -194,7 +195,7 @@ async def test_user_client_can_submit_task(result, context: AbstractContextManag with context: await user_client.can_submit_task( - TaskRequest(name="foo", params={}, instrument_session="cm12345-1") + TaskRequest(name="foo", params=TaskParams(), instrument_session="cm12345-1") ) opa.require_submit_task.assert_called_once_with("cm12345-1", "foo_bar") diff --git a/tests/unit_tests/service/test_interface.py b/tests/unit_tests/service/test_interface.py index 892c5ad2ea..94617381c0 100644 --- a/tests/unit_tests/service/test_interface.py +++ b/tests/unit_tests/service/test_interface.py @@ -46,7 +46,7 @@ WorkerEvent, WorkerState, ) -from blueapi.worker.task import Task +from blueapi.worker.task import Task, TaskParams from blueapi.worker.task_worker import TrackableTask FAKE_INSTRUMENT_SESSION = "cm12345-1" @@ -385,7 +385,7 @@ def test_get_task_by_id( request_id=ANY, task=Task( name="my_plan", - params={}, + params=TaskParams(), metadata=expected_metadata, ), is_complete=False, @@ -417,7 +417,7 @@ def test_submit_task_inserts_metadata(context_mock: MagicMock): request_id=ANY, task=Task( name="my_plan", - params={}, + params=TaskParams(), metadata=expected_metadata, ), is_complete=False, diff --git a/tests/unit_tests/service/test_protocol.py b/tests/unit_tests/service/test_protocol.py index 9178162e18..08a0e39130 100644 --- a/tests/unit_tests/service/test_protocol.py +++ b/tests/unit_tests/service/test_protocol.py @@ -15,6 +15,7 @@ Resume, Submit, ) +from blueapi.worker import TaskParams @pytest.mark.parametrize( @@ -29,7 +30,9 @@ } }""", Submit( - task=TaskRequest(name="foo", params={}, instrument_session="cm12345-1") + task=TaskRequest( + name="foo", params=TaskParams(), instrument_session="cm12345-1" + ) ), ), ('{"kind": "pause"}', Pause()), From 29b05e4173b444e293d16c9404853f1b0aea7ec3 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 9 Sep 2026 16:35:54 +0000 Subject: [PATCH 08/31] fix SUBMIT_REQUEST --- tests/unit_tests/service/test_rest_api.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit_tests/service/test_rest_api.py b/tests/unit_tests/service/test_rest_api.py index 712b246f5c..9da3a754cb 100644 --- a/tests/unit_tests/service/test_rest_api.py +++ b/tests/unit_tests/service/test_rest_api.py @@ -62,7 +62,7 @@ class MockCountModel(BaseModel): ... "kind": "submit", "task": { "name": "foo", - "params": {"one": "two"}, + "params": {"args": [], "kwargs": {"one": "two"}}, "instrument_session": "cm12345-1", }, } From adfa659bac4a62b1441286d5a0eacb3623fb8bb1 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 9 Sep 2026 16:51:51 +0000 Subject: [PATCH 09/31] Correct doc string --- src/blueapi/worker/task.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/src/blueapi/worker/task.py b/src/blueapi/worker/task.py index 4c64698930..a7291ce684 100644 --- a/src/blueapi/worker/task.py +++ b/src/blueapi/worker/task.py @@ -34,15 +34,13 @@ def prepare_params( self, ctx: BlueskyContext ) -> tuple[list[Any], Mapping[str, Any]]: """ - Checks plan parameters against context + Checks the configured plan parameters against context Args: ctx: Context holding plans and devices - plan: Plan object including schema - params: Parameter values to be validated against schema Returns: - Mapping[str, Any]: _description_ + tuple[list[Any], Mapping[str, Any]]: The prepared parameters for the plan. """ plan = ctx.plans[self.name] func = ctx.plan_functions[self.name] From e26f0b018f51b16a9fce035e4a142048c705770f Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Mon, 12 Oct 2026 20:46:55 +0000 Subject: [PATCH 10/31] Add backwards compatability --- docs/reference/openapi.yaml | 8 +-- src/blueapi/cli/cli.py | 66 ++++++++++++------- src/blueapi/client/client.py | 18 ++--- src/blueapi/client/rest.py | 9 ++- src/blueapi/service/interface.py | 5 ++ src/blueapi/service/main.py | 10 +-- src/blueapi/service/model.py | 41 ++++++++++-- src/blueapi/service/protocol.py | 4 +- src/blueapi/worker/__init__.py | 5 +- src/blueapi/worker/task.py | 14 +++- tests/system_tests/test_blueapi_system.py | 62 ++++++++--------- tests/unit_tests/cli/test_cli.py | 30 ++++++--- tests/unit_tests/client/test_client.py | 58 +++++++++------- tests/unit_tests/client/test_rest.py | 18 ++--- .../unit_tests/service/test_authorization.py | 8 ++- tests/unit_tests/service/test_interface.py | 35 +++++++--- tests/unit_tests/service/test_protocol.py | 10 +-- tests/unit_tests/service/test_rest_api.py | 22 +++---- tests/unit_tests/worker/test_task_worker.py | 34 ++++++---- 19 files changed, 289 insertions(+), 168 deletions(-) diff --git a/docs/reference/openapi.yaml b/docs/reference/openapi.yaml index e0cd8282df..9747fc0968 100644 --- a/docs/reference/openapi.yaml +++ b/docs/reference/openapi.yaml @@ -274,7 +274,7 @@ components: - message title: TaskError type: object - TaskRequest: + TaskRequestV2: additionalProperties: false description: Request to run a task with related info properties: @@ -294,7 +294,7 @@ components: required: - name - instrument_session - title: TaskRequest + title: TaskRequestV2 type: object TaskResponse: additionalProperties: false @@ -641,7 +641,7 @@ paths: content: application/json: schema: - $ref: '#/components/schemas/TaskRequest' + $ref: '#/components/schemas/TaskRequestV2' examples: - instrument_session: cm12345-1 name: count @@ -1042,7 +1042,7 @@ paths: content: application/json: schema: - $ref: '#/components/schemas/TaskRequest' + $ref: '#/components/schemas/TaskRequestV2' examples: - instrument_session: cm12345-1 name: count diff --git a/src/blueapi/cli/cli.py b/src/blueapi/cli/cli.py index a3a9e56821..9f4deb4d21 100644 --- a/src/blueapi/cli/cli.py +++ b/src/blueapi/cli/cli.py @@ -31,15 +31,21 @@ UnauthorisedAccessError, UnknownPlanError, ) -from blueapi.config import ( - ApplicationConfig, - ConfigLoader, -) +from blueapi.config import ApplicationConfig, ConfigLoader from blueapi.core import OTLP_EXPORT_ENABLED, DataEvent from blueapi.log import set_up_logging from blueapi.service.authentication import SessionCacheManager, SessionManager -from blueapi.service.model import DeviceResponse, PlanResponse, SourceInfo, TaskRequest -from blueapi.worker import ProgressEvent, TaskParams, WorkerEvent +from blueapi.service.model import ( + DeviceResponse, + PlanResponse, + SourceInfo, + TaskRequestV2, +) +from blueapi.worker import ( + ProgressEvent, + TaskParametersV2, + WorkerEvent, +) from blueapi.worker.event import TaskError, TaskResult from .scratch import setup_scratch @@ -47,33 +53,48 @@ LOGGER = logging.getLogger(__name__) -TaskParameters = dict[str, Any] - class ParametersType(ParamType): - """CLI input parameter to accept a JSON object as an argument""" + """CLI input parameter accepting V1 or V2 JSON parameters.""" - name = "TaskParameters" + name = "parameters" def convert( self, value: str | dict[str, Any] | None, param: Parameter | None, ctx: Context | None, - ) -> TaskParameters: + ) -> TaskParametersV2: + if value is None: + return TaskParametersV2() + if isinstance(value, str): try: - params = json.loads(value) - if is_str_dict(params): - return params - self.fail("Parameters must be a JSON object with string keys") + value = json.loads(value) except json.JSONDecodeError as jde: - self.fail(f"Parameters are not valid JSON: {jde}") - else: - return super().convert(value, param, ctx) + self.fail(f"Parameters are not valid JSON: {jde}", param, ctx) + + if not isinstance(value, dict) or not all( + isinstance(key, str) for key in value + ): + self.fail( + "Parameters must be a JSON object with string keys", + param, + ctx, + ) + + # V2 parameters have the shape {"args": [...], "kwargs": {...}}. + if value and set(value).issubset({"args", "kwargs"}): + try: + return TaskParametersV2.model_validate(value) + except ValueError as exc: + self.fail(f"Invalid V2 parameters: {exc}", param, ctx) + # Otherwise, interpret the object as V1 keyword arguments. + return TaskParametersV2.from_v1(value) -def is_str_dict(val: Any) -> TypeGuard[TaskParameters]: + +def is_str_dict(val: Any) -> TypeGuard[TaskParametersV2]: return isinstance(val, dict) and all(isinstance(k, str) for k in val) @@ -363,7 +384,7 @@ def run_plan( foreground: bool, ws: bool, instrument_session: str, - parameters: TaskParameters, + parameters: TaskParametersV2, ) -> None: """Run a plan with parameters @@ -371,11 +392,10 @@ def run_plan( configuration or the experimental --ws flag is required. Without stomp configuration, '--bg' can be used to start a plan in the background. """ - client = cast(BlueapiClient, obj["client"]) - task = TaskRequest( + task = TaskRequestV2( name=name, - params=TaskParams(kwargs=parameters), + params=parameters, instrument_session=instrument_session, ) diff --git a/src/blueapi/client/client.py b/src/blueapi/client/client.py index 3d77a93daa..6e1e895323 100644 --- a/src/blueapi/client/client.py +++ b/src/blueapi/client/client.py @@ -31,8 +31,8 @@ PlanResponse, PythonEnvironmentResponse, SourceInfo, - TaskParams, - TaskRequest, + TaskParametersV2, + TaskRequestV2, TaskResponse, TasksListResponse, WorkerTask, @@ -158,7 +158,7 @@ def __init__(self, name, model: PlanModel, client: "BlueapiClient"): self.__doc__ = model.description def __call__(self, *args, **kwargs) -> Any: - req = TaskRequest( + req = TaskRequestV2( name=self.name, params=self._build_args(*args, **kwargs), instrument_session=self._client.instrument_session, @@ -181,7 +181,7 @@ def properties(self) -> dict[str, Any]: def required(self) -> list[str]: return self.model.parameter_schema.get("required", []) - def _build_args(self, *args, **kwargs) -> TaskParams: + def _build_args(self, *args, **kwargs) -> TaskParametersV2: log.info( "Building args for %s, using %s and %s", "[" + ",".join(self.properties) + "]", @@ -208,7 +208,7 @@ def _build_args(self, *args, **kwargs) -> TaskParams: if missing := set(self.required) - supplied: raise TypeError(f"Missing argument(s) for {missing}") - return TaskParams(args=args, kwargs=kwargs) + return TaskParametersV2(args=args, kwargs=kwargs) def __repr__(self) -> str: required = set(self.required) @@ -489,7 +489,7 @@ def get_active_task(self) -> WorkerTask: @start_as_current_span(TRACER, "request") def run_blocking( - self, request: TaskRequest, on_event: OnAnyEvent | None = None + self, request: TaskRequestV2, on_event: OnAnyEvent | None = None ) -> TaskStatus: for event in self._rest.run_blocking(request): if on_event is not None: @@ -511,7 +511,7 @@ def run_blocking( @start_as_current_span(TRACER, "task", "timeout") def run_task( self, - task: TaskRequest, + task: TaskRequestV2, on_event: OnAnyEvent | None = None, timeout: float | None = None, ) -> TaskStatus: @@ -579,7 +579,7 @@ def inner_on_event(event: AnyEvent, ctx: MessageContext) -> None: return complete.result(timeout=timeout) @start_as_current_span(TRACER, "task") - def create_and_start_task(self, task: TaskRequest) -> TaskResponse: + def create_and_start_task(self, task: TaskRequestV2) -> TaskResponse: """ Create a new task and instruct the worker to start it immediately. @@ -605,7 +605,7 @@ def create_and_start_task(self, task: TaskRequest) -> TaskResponse: @start_as_current_span(TRACER, "task") @deprecated("rest client") - def create_task(self, task: TaskRequest) -> TaskResponse: + def create_task(self, task: TaskRequestV2) -> TaskResponse: """ Create a new task, does not start execution diff --git a/src/blueapi/client/rest.py b/src/blueapi/client/rest.py index 6e03040c47..efa1ff04c5 100644 --- a/src/blueapi/client/rest.py +++ b/src/blueapi/client/rest.py @@ -30,6 +30,8 @@ PythonEnvironmentResponse, SourceInfo, TaskRequest, + TaskRequestV1, + TaskRequestV2, TaskResponse, TasksListResponse, WorkerTask, @@ -253,6 +255,8 @@ def get_active_task(self) -> WorkerTask: return self._request_and_deserialize("/worker/task", WorkerTask) def create_task(self, task: TaskRequest) -> TaskResponse: + if isinstance(task, TaskRequestV1): + task = TaskRequestV2.from_v1(task) return self._request_and_deserialize( "/tasks", TaskResponse, @@ -359,6 +363,9 @@ def _request_and_deserialize( def run_blocking( self, req: TaskRequest ) -> Iterable[DataEvent | WorkerEvent | ProgressEvent]: + if isinstance(req, TaskRequestV1): + req = TaskRequestV2.from_v1(req) + url = self._config.ws_address.unicode_string().rstrip("/") + "/api/v2/run_plan" headers = get_context_propagator() if self.session_manager: @@ -430,7 +437,7 @@ class ServiceUnavailableError(Exception): def _task_model_fallback(obj: Any) -> Any: - """Fallback method for serializing TaskRequests""" + """Fallback method for serializing TaskRequest""" if isinstance(obj, client.DeviceRef): return obj.name raise PydanticSerializationError(f"Object of type {type(obj)} not serializable") diff --git a/src/blueapi/service/interface.py b/src/blueapi/service/interface.py index 6c0c5befbd..bc9a581849 100644 --- a/src/blueapi/service/interface.py +++ b/src/blueapi/service/interface.py @@ -23,6 +23,8 @@ PythonEnvironmentResponse, SourceInfo, TaskRequest, + TaskRequestV1, + TaskRequestV2, WorkerTask, ) from blueapi.utils.serialization import access_blob @@ -156,6 +158,9 @@ def submit_task( task_request: TaskRequest, metadata: dict[str, Any] | None = None ) -> str: """Submit a task to be run on begin_task""" + if isinstance(task_request, TaskRequestV1): + task_request = TaskRequestV2.from_v1(task_request) + # Can't default arg to mutable data structure: if metadata is None: metadata = {} diff --git a/src/blueapi/service/main.py b/src/blueapi/service/main.py index 81b7fd1cc7..a65d5e0569 100644 --- a/src/blueapi/service/main.py +++ b/src/blueapi/service/main.py @@ -55,7 +55,7 @@ Unauthorized, Update, ) -from blueapi.worker import TaskParams, TrackableTask, WorkerState +from blueapi.worker import TaskParametersV2, TrackableTask, WorkerState from blueapi.worker.event import ProgressEvent, TaskStatusEnum, WorkerEvent from blueapi.worker.worker_errors import WorkerBusyError @@ -77,7 +77,7 @@ PythonEnvironmentResponse, SourceInfo, StateChangeRequest, - TaskRequest, + TaskRequestV2, TaskResponse, TasksListResponse, WorkerTask, @@ -318,9 +318,9 @@ def get_device_by_name( return runner.run(interface.get_device, name) -example_task_request = TaskRequest( +example_task_request = TaskRequestV2( name="count", - params=TaskParams(kwargs={"detectors": ["x"]}), + params=TaskParametersV2(kwargs={"detectors": ["x"]}), instrument_session="cm12345-1", ) @@ -337,7 +337,7 @@ def get_device_by_name( def submit_task( request: Request, response: Response, - task_request: Annotated[TaskRequest, Body(..., examples=[example_task_request])], + task_request: Annotated[TaskRequestV2, Body(..., examples=[example_task_request])], _: Annotated[None, Depends(submit_permission)], runner: Annotated[WorkerDispatcher, Depends(_runner)], fedid: Fedid, diff --git a/src/blueapi/service/model.py b/src/blueapi/service/model.py index 7d7aaabc1d..3f133f808d 100644 --- a/src/blueapi/service/model.py +++ b/src/blueapi/service/model.py @@ -1,5 +1,5 @@ import uuid -from collections.abc import Iterable +from collections.abc import Iterable, Mapping from enum import StrEnum from typing import Annotated, Any @@ -11,7 +11,7 @@ from blueapi.core import BLUESKY_PROTOCOLS, Device, Plan from blueapi.core.context import generic_bounds from blueapi.utils import BlueapiBaseModel -from blueapi.worker import TaskParams, WorkerState +from blueapi.worker import TaskParametersV2, WorkerState from blueapi.worker.task_worker import TaskWorker, TrackableTask _UNKNOWN_NAME = "UNKNOWN" @@ -58,19 +58,48 @@ class TasksListResponse(BlueapiBaseModel): tasks: list[TrackableTask] = Field(description="List of tasks") -class TaskRequest(BlueapiBaseModel): +TaskParametersV2V1 = Mapping[str, Any] + + +class TaskRequestV1(BlueapiBaseModel): + """ + Request to run a task with related info. Supports **kwargs only. + """ + + name: str = Field(description="Name of plan to run") + params: TaskParametersV2V1 = Field( + description="Values for parameters to plan, if any", default_factory=dict + ) + instrument_session: str = Field( + description="Instrument session associated with this task", + ) + + +class TaskRequestV2(BlueapiBaseModel): """ - Request to run a task with related info + Request to run a task with related info. Supports *args and **kwargs. """ name: str = Field(description="Name of plan to run") - params: TaskParams = Field( - description="Values for parameters to plan, if any", default_factory=TaskParams + params: TaskParametersV2 = Field( + description="Values for parameters to plan, if any", + default_factory=TaskParametersV2, ) instrument_session: str = Field( description="Instrument session associated with this task", ) + @classmethod + def from_v1(cls, task_v1: TaskRequestV1) -> "TaskRequestV2": + return cls( + name=task_v1.name, + instrument_session=task_v1.instrument_session, + params=TaskParametersV2(kwargs=task_v1.params), + ) + + +TaskRequest = TaskRequestV1 | TaskRequestV2 + class DeviceRequest(BlueapiBaseModel): """ diff --git a/src/blueapi/service/protocol.py b/src/blueapi/service/protocol.py index 8af5819133..7c20fa4fbe 100644 --- a/src/blueapi/service/protocol.py +++ b/src/blueapi/service/protocol.py @@ -20,7 +20,7 @@ from pydantic import BaseModel, Field, TypeAdapter, ValidationError from blueapi.core.bluesky_types import DataEvent -from blueapi.service.model import TaskRequest +from blueapi.service.model import TaskRequestV2 from blueapi.worker.event import ProgressEvent, WorkerEvent UNKNOWN_PLAN = 4001 @@ -37,7 +37,7 @@ class ArgumentError(BaseModel): class Submit(BaseModel): kind: Literal["submit"] = "submit" - task: TaskRequest + task: TaskRequestV2 class Pause(BaseModel): diff --git a/src/blueapi/worker/__init__.py b/src/blueapi/worker/__init__.py index 5002da8a8c..6c0aaacb41 100644 --- a/src/blueapi/worker/__init__.py +++ b/src/blueapi/worker/__init__.py @@ -1,12 +1,13 @@ from .event import ProgressEvent, StatusView, TaskStatus, WorkerEvent, WorkerState -from .task import Task, TaskParams +from .task import Task, TaskParametersV1, TaskParametersV2 from .task_worker import TaskWorker, TrackableTask from .worker_errors import WorkerAlreadyStartedError, WorkerBusyError __all__ = [ "TaskWorker", "Task", - "TaskParams", + "TaskParametersV1", + "TaskParametersV2", "WorkerEvent", "WorkerState", "StatusView", diff --git a/src/blueapi/worker/task.py b/src/blueapi/worker/task.py index a7291ce684..0284b14c93 100644 --- a/src/blueapi/worker/task.py +++ b/src/blueapi/worker/task.py @@ -11,10 +11,17 @@ LOGGER = logging.getLogger(__name__) -class TaskParams(BlueapiBaseModel): +TaskParametersV1 = Mapping[str, Any] + + +class TaskParametersV2(BlueapiBaseModel): args: Sequence[Any] = Field(default_factory=lambda: []) kwargs: Mapping[str, Any] = Field(default_factory=dict) + @classmethod + def from_v1(cls, param: TaskParametersV1): + return cls(kwargs=param) + class Task(BlueapiBaseModel): """ @@ -22,8 +29,9 @@ class Task(BlueapiBaseModel): """ name: str = Field(description="Name of plan to run") - params: TaskParams = Field( - description="Values for parameters to plan, if any", default_factory=TaskParams + params: TaskParametersV2 = Field( + description="Values for parameters to plan, if any", + default_factory=TaskParametersV2, ) metadata: dict[str, Any] = Field( description="Any metadata to apply to all runs within this task", diff --git a/tests/system_tests/test_blueapi_system.py b/tests/system_tests/test_blueapi_system.py index 55237850e7..c3b564f55b 100644 --- a/tests/system_tests/test_blueapi_system.py +++ b/tests/system_tests/test_blueapi_system.py @@ -32,8 +32,8 @@ from blueapi.service.model import ( DeviceResponse, PlanResponse, - TaskParams, - TaskRequest, + TaskParametersV2, + TaskRequestV2, TaskResponse, WorkerTask, ) @@ -120,10 +120,10 @@ def scan_id(local_numtracker) -> int: def task_factory( user: ValidUser, instrument_session: str | None, time: float = 0.0 -) -> TaskRequest: - return TaskRequest( +) -> TaskRequestV2: + return TaskRequestV2( name="sleep", - params=TaskParams(kwargs={"time": time}), + params=TaskParametersV2(kwargs={"time": time}), instrument_session=instrument_session if instrument_session else VALID_INSTRUMENT_SESSION[user], @@ -131,12 +131,12 @@ def task_factory( @pytest.fixture -def small_task(user: ValidUser, instrument_session: str | None) -> TaskRequest: +def small_task(user: ValidUser, instrument_session: str | None) -> TaskRequestV2: return task_factory(user, instrument_session) @pytest.fixture -def long_task(user: ValidUser, instrument_session: str | None) -> TaskRequest: +def long_task(user: ValidUser, instrument_session: str | None) -> TaskRequestV2: return task_factory(user, instrument_session, time=1.0) @@ -340,14 +340,14 @@ def test_client_non_existent_device(client: BlueapiClient): def test_create_task_and_delete_task_by_id( - rest_client: BlueapiRestClient, small_task: TaskRequest + rest_client: BlueapiRestClient, small_task: TaskRequestV2 ): create_task = rest_client.create_task(small_task) rest_client.clear_task(create_task.task_id) def test_task_metadata_propagated( - rest_client: BlueapiRestClient, small_task: TaskRequest + rest_client: BlueapiRestClient, small_task: TaskRequestV2 ): response = rest_client.create_task(small_task) trackable_task = rest_client.get_task(response.task_id) @@ -364,16 +364,16 @@ def test_task_metadata_propagated( def test_create_task_validation_error(rest_client: BlueapiRestClient): with pytest.raises(BlueskyRequestError, match="Internal Server Error"): rest_client.create_task( - TaskRequest( + TaskRequestV2( name="Not-exists", - params=TaskParams(kwargs={"Not-exists": 0.0}), + params=TaskParametersV2(kwargs={"Not-exists": 0.0}), instrument_session="Not-exists", ) ) def test_get_all_tasks( - rest_client: BlueapiRestClient, small_task: TaskRequest, long_task: TaskRequest + rest_client: BlueapiRestClient, small_task: TaskRequestV2, long_task: TaskRequestV2 ): created_tasks: list[TaskResponse] = [] for task in [small_task, long_task]: @@ -390,7 +390,7 @@ def test_get_all_tasks( rest_client.clear_task(task_id) -def test_get_task_by_id(rest_client: BlueapiRestClient, small_task: TaskRequest): +def test_get_task_by_id(rest_client: BlueapiRestClient, small_task: TaskRequestV2): created_task = rest_client.create_task(small_task) get_task = rest_client.get_task(created_task.task_id) @@ -414,7 +414,7 @@ def test_delete_non_existent_task(rest_client: BlueapiRestClient): rest_client.clear_task("Not-exists") -def test_put_worker_task(rest_client: BlueapiRestClient, small_task: TaskRequest): +def test_put_worker_task(rest_client: BlueapiRestClient, small_task: TaskRequestV2): created_task = rest_client.create_task(small_task) rest_client.update_worker_task(WorkerTask(task_id=created_task.task_id)) active_task = rest_client.get_active_task() @@ -423,7 +423,7 @@ def test_put_worker_task(rest_client: BlueapiRestClient, small_task: TaskRequest def test_put_worker_task_fails_if_not_idle( - rest_client: BlueapiRestClient, small_task: TaskRequest, long_task: TaskRequest + rest_client: BlueapiRestClient, small_task: TaskRequestV2, long_task: TaskRequestV2 ): small_task_id = rest_client.create_task(small_task).task_id long_task_id = rest_client.create_task(long_task).task_id @@ -453,7 +453,7 @@ def test_set_state_transition_error(client: BlueapiClient): assert "Cannot transition from IDLE to PAUSED" in exception.value.args[1] -def test_get_task_by_status(rest_client: BlueapiRestClient, small_task: TaskRequest): +def test_get_task_by_status(rest_client: BlueapiRestClient, small_task: TaskRequestV2): task_1 = rest_client.create_task(small_task) task_2 = rest_client.create_task(small_task) task_by_pending = rest_client.get_all_tasks() @@ -484,7 +484,9 @@ def test_get_task_by_status(rest_client: BlueapiRestClient, small_task: TaskRequ rest_client.clear_task(task_id=task_2.task_id) -def test_progress_with_stomp(client_with_stomp: BlueapiClient, small_task: TaskRequest): +def test_progress_with_stomp( + client_with_stomp: BlueapiClient, small_task: TaskRequestV2 +): all_events: list[AnyEvent] = [] def on_event(event: AnyEvent): @@ -541,17 +543,17 @@ def test_delete_current_environment(client: BlueapiClient): "task,user", [ ( - TaskRequest( + TaskRequestV2( name="count", - params=TaskParams(kwargs={"detectors": ["det"], "num": 3}), + params=TaskParametersV2(kwargs={"detectors": ["det"], "num": 3}), instrument_session=VALID_INSTRUMENT_SESSION[User.alice], ), User.alice, ), ( - TaskRequest( + TaskRequestV2( name="spec_scan", - params=TaskParams( + params=TaskParametersV2( kwargs={ "detectors": ["det"], "spec": { @@ -583,7 +585,7 @@ def test_delete_current_environment(client: BlueapiClient): @pytest.mark.parametrize("run_method", ["run_task", "run_blocking"]) def test_plan_runs( client_with_stomp: BlueapiClient, - task: TaskRequest, + task: TaskRequestV2, scan_id: int, user: ValidUser, run_method: str, @@ -639,9 +641,9 @@ def on_event(event: AnyEvent) -> None: @pytest.mark.parametrize( "task", [ - TaskRequest( + TaskRequestV2( name="set_absolute", - params=TaskParams( + params=TaskParametersV2( kwargs={ "movable": "stage.x", "value": 1.0, @@ -651,7 +653,7 @@ def on_event(event: AnyEvent) -> None: ), ], ) -def test_stub_runs(client_with_stomp: BlueapiClient, task: TaskRequest): +def test_stub_runs(client_with_stomp: BlueapiClient, task: TaskRequestV2): final_event = client_with_stomp.run_task(task) assert isinstance(final_event.result, TaskResult) assert final_event.task_complete @@ -666,9 +668,9 @@ def test_task_submission_after_invalid_task(client_with_stomp: BlueapiClient): # ...but should leave the serve in a state where it can still run tasks res = client_with_stomp.run_task( - TaskRequest( + TaskRequestV2( name="count", - params=TaskParams( + params=TaskParametersV2( kwargs={ "detectors": ["det"], } @@ -693,7 +695,7 @@ def test_task_submission_after_invalid_task(client_with_stomp: BlueapiClient): @pytest.mark.parametrize("method_name", ["create_task", "run_blocking"]) def test_run_task_without_authz( client: BlueapiClient, - small_task: TaskRequest, + small_task: TaskRequestV2, user: ValidUser, instrument_session: str, method_name: str, @@ -718,7 +720,7 @@ def test_run_task_without_authz( ) def test_create_task_authorization( client: BlueapiClient, - small_task: TaskRequest, + small_task: TaskRequestV2, user: ValidUser, instrument_session: str, ): @@ -886,7 +888,7 @@ def test_admin_can_abort_any_task( def test_run_blocking_requires_auth( - client_without_auth: BlueapiClient, small_task: TaskRequest + client_without_auth: BlueapiClient, small_task: TaskRequestV2 ): with pytest.raises(UnauthorisedAccessError): client_without_auth.run_blocking(small_task) diff --git a/tests/unit_tests/cli/test_cli.py b/tests/unit_tests/cli/test_cli.py index d491d45d7d..3d9227ae0d 100644 --- a/tests/unit_tests/cli/test_cli.py +++ b/tests/unit_tests/cli/test_cli.py @@ -47,10 +47,10 @@ PlanModel, PlanResponse, PythonEnvironmentResponse, - TaskRequest, + TaskRequestV2, TaskResponse, ) -from blueapi.worker import TaskParams +from blueapi.worker import TaskParametersV2 from blueapi.worker.event import ( ProgressEvent, TaskError, @@ -261,9 +261,17 @@ def test_submit_plan_without_stomp(runner: CliRunner): ) +@pytest.mark.parametrize( + "params", + [ + {"time": 3}, + {"args": [], "kwargs": {"time": 3}}, + ], + ids=["v1", "v2"], +) @patch("blueapi.client.client.StompClient") @responses.activate -def test_run_plan(stomp_client: StompClient, runner: CliRunner): +def test_run_plan(stomp_client: StompClient, runner: CliRunner, params): task_id = "abcd-1234" submit_response = responses.post( url="http://a.fake.host:12345/tasks", @@ -340,11 +348,11 @@ def mock_events(topic: MessageTopic, callback: Callable[[Any, Any], Any]): "-i", "cm12345-1", "sleep", - '{"time": 3}', + json.dumps(params), ], ) - assert result.exit_code == 0 + assert result.exit_code == 0, result.output assert submit_response.call_count == 1 assert run_response.call_count == 1 @@ -387,7 +395,9 @@ def test_run_plan_feedback( ) bc.add_callback.assert_called_once() bc.run_task.assert_called_once_with( - TaskRequest(name="name", params=TaskParams(), instrument_session="cm12345-1"), + TaskRequestV2( + name="name", params=TaskParametersV2(), instrument_session="cm12345-1" + ), ) assert res.exit_code == 0 assert res.stdout == message @@ -502,9 +512,9 @@ def test_can_pass_an_instrument_session_with_an_environment_variable( ) assert result.exit_code == 0, result.output mock_create_task.assert_called_once_with( - TaskRequest( + TaskRequestV2( name="sleep", - params=TaskParams(kwargs={"time": 5.0}), + params=TaskParametersV2(kwargs={"time": 5.0}), instrument_session="cm12345-1", ) ) @@ -1490,6 +1500,8 @@ def test_run_ws_runs_blocking_plan(mock_client: Mock, runner: CliRunner): bc.add_callback.assert_called_once() bc.run_task.assert_not_called() bc.run_blocking.assert_called_once_with( - TaskRequest(name="name", params=TaskParams(), instrument_session="cm12345-1"), + TaskRequestV2( + name="name", params=TaskParametersV2(), instrument_session="cm12345-1" + ), ) assert res.exit_code == 0 diff --git a/tests/unit_tests/client/test_client.py b/tests/unit_tests/client/test_client.py index 13502270cc..8175cc3d0e 100644 --- a/tests/unit_tests/client/test_client.py +++ b/tests/unit_tests/client/test_client.py @@ -37,7 +37,7 @@ PlanModel, PlanResponse, ProtocolInfo, - TaskRequest, + TaskRequestV2, TaskResponse, TasksListResponse, WorkerTask, @@ -45,7 +45,7 @@ from blueapi.worker import ( ProgressEvent, Task, - TaskParams, + TaskParametersV2, TrackableTask, WorkerEvent, WorkerState, @@ -79,7 +79,7 @@ ] ) DEVICE = DeviceModel(name="foo", protocols=[]) -TASK = TrackableTask(task_id="foo", task=Task(name="bar", params=TaskParams())) +TASK = TrackableTask(task_id="foo", task=Task(name="bar", params=TaskParametersV2())) TASKS = TasksListResponse(tasks=[TASK]) ACTIVE_TASK = WorkerTask(task_id="bar") ENVIRONMENT_ID = uuid.uuid4() @@ -238,10 +238,10 @@ def test_create_and_start_task_calls_both_creating_and_starting_endpoints( mock_rest.create_task.return_value = TaskResponse(task_id="baz") mock_rest.update_worker_task.return_value = TaskResponse(task_id="baz") client.create_and_start_task( - TaskRequest(name="baz", instrument_session="cm12345-1") + TaskRequestV2(name="baz", instrument_session="cm12345-1") ) mock_rest.create_task.assert_called_once_with( - TaskRequest(name="baz", instrument_session="cm12345-1") + TaskRequestV2(name="baz", instrument_session="cm12345-1") ) mock_rest.update_worker_task.assert_called_once_with(WorkerTask(task_id="baz")) @@ -253,7 +253,7 @@ def test_create_and_start_task_fails_if_task_creation_fails( mock_rest.create_task.side_effect = BlueskyRemoteControlError("No can do") with pytest.raises(BlueskyRemoteControlError): client.create_and_start_task( - TaskRequest(name="baz", instrument_session="cm12345-1") + TaskRequestV2(name="baz", instrument_session="cm12345-1") ) @@ -265,7 +265,7 @@ def test_create_and_start_task_fails_if_task_id_is_wrong( mock_rest.update_worker_task.return_value = TaskResponse(task_id="bar") with pytest.raises(BlueskyRemoteControlError): client.create_and_start_task( - TaskRequest(name="baz", instrument_session="cm12345-1") + TaskRequestV2(name="baz", instrument_session="cm12345-1") ) @@ -277,7 +277,7 @@ def test_create_and_start_task_fails_if_task_start_fails( mock_rest.update_worker_task.side_effect = BlueskyRemoteControlError("No can do") with pytest.raises(BlueskyRemoteControlError): client.create_and_start_task( - TaskRequest(name="baz", instrument_session="cm12345-1") + TaskRequestV2(name="baz", instrument_session="cm12345-1") ) @@ -446,7 +446,7 @@ def test_cannot_run_task_without_message_bus(client: BlueapiClient): MissingStompConfigurationError, match="Stomp configuration required to run plans is missing or disabled", ): - client.run_task(TaskRequest(name="foo", instrument_session="cm12345-1")) + client.run_task(TaskRequestV2(name="foo", instrument_session="cm12345-1")) def test_run_task_sets_up_control( @@ -460,9 +460,11 @@ def test_run_task_sets_up_control( ctx.correlation_id = "foo" mock_events.subscribe_to_all_events = lambda on_event: on_event(COMPLETE_EVENT, ctx) - client_with_events.run_task(TaskRequest(name="foo", instrument_session="cm12345-1")) + client_with_events.run_task( + TaskRequestV2(name="foo", instrument_session="cm12345-1") + ) mock_rest.create_task.assert_called_once_with( - TaskRequest(name="foo", instrument_session="cm12345-1") + TaskRequestV2(name="foo", instrument_session="cm12345-1") ) mock_rest.update_worker_task.assert_called_once_with(WorkerTask(task_id="foo")) @@ -481,7 +483,7 @@ def test_run_task_fails_on_failing_event( on_event = Mock() outcome = client_with_events.run_task( - TaskRequest(name="foo", instrument_session="cm12345-1"), + TaskRequestV2(name="foo", instrument_session="cm12345-1"), on_event=on_event, ) assert outcome.task_failed @@ -529,7 +531,8 @@ def callback(on_event: Callable[[AnyEvent, MessageContext], None]): mock_on_event = Mock() client_with_events.run_task( - TaskRequest(name="foo", instrument_session="cm12345-1"), on_event=mock_on_event + TaskRequestV2(name="foo", instrument_session="cm12345-1"), + on_event=mock_on_event, ) assert mock_on_event.mock_calls == [call(test_event), call(COMPLETE_EVENT)] @@ -571,7 +574,8 @@ def callback(on_event: Callable[[AnyEvent, MessageContext], None]): mock_on_event = Mock() client_with_events.run_task( - TaskRequest(name="foo", instrument_session="cm12345-1"), on_event=mock_on_event + TaskRequestV2(name="foo", instrument_session="cm12345-1"), + on_event=mock_on_event, ) mock_on_event.assert_called_once_with(COMPLETE_EVENT) @@ -659,7 +663,7 @@ def test_create_and_start_task_span_ok( mock_rest.update_worker_task.return_value = TaskResponse(task_id="baz") with asserting_span_exporter(exporter, "create_and_start_task", "task"): client.create_and_start_task( - TaskRequest(name="baz", instrument_session="cm12345-1") + TaskRequestV2(name="baz", instrument_session="cm12345-1") ) @@ -724,7 +728,7 @@ def test_cannot_run_task_span_ok( match="Stomp configuration required to run plans is missing or disabled", ): with asserting_span_exporter(exporter, "grun_task"): - client.run_task(TaskRequest(name="foo", instrument_session="cm12345-1")) + client.run_task(TaskRequestV2(name="foo", instrument_session="cm12345-1")) def test_instrument_session_required(client): @@ -910,31 +914,31 @@ def test_plan_empty_fallback_help_text(client): p( (1,), {}, - TaskParams(args=(1,), kwargs={}), + TaskParametersV2(args=(1,), kwargs={}), id="required_as_positional", ), p( (), {"one": 7}, - TaskParams(args=(), kwargs={"one": 7}), + TaskParametersV2(args=(), kwargs={"one": 7}), id="required_as_keyword", ), p( (1,), {"two": 23}, - TaskParams(args=(1,), kwargs={"two": 23}), + TaskParametersV2(args=(1,), kwargs={"two": 23}), id="all_as_mixed_args_kwargs", ), p( (1, 2), {}, - TaskParams(args=(1, 2), kwargs={}), + TaskParametersV2(args=(1, 2), kwargs={}), id="all_as_positional", ), p( (), {"one": 21, "two": 42}, - TaskParams(args=(), kwargs={"one": 21, "two": 42}), + TaskParametersV2(args=(), kwargs={"one": 21, "two": 42}), id="all_as_keyword", ), ], @@ -950,7 +954,7 @@ def test_plan_param_mapping(args, kwargs, params): plan(*args, **kwargs) client.run_task.assert_called_once_with( - TaskRequest(name="foobar", instrument_session="cm12345-1", params=params) + TaskRequestV2(name="foobar", instrument_session="cm12345-1", params=params) ) @@ -1029,7 +1033,9 @@ def subscribe(on_event: Callable[[AnyEvent, MessageContext], None]): mock_events.subscribe_to_all_events = subscribe # type: ignore - client_with_events.run_task(TaskRequest(name="foo", instrument_session="cm12345-1")) + client_with_events.run_task( + TaskRequestV2(name="foo", instrument_session="cm12345-1") + ) assert callback.mock_calls == [call(test_event), call(COMPLETE_EVENT)] @@ -1057,7 +1063,9 @@ def subscribe(on_event: Callable[[AnyEvent, MessageContext], None]): mock_events.subscribe_to_all_events = subscribe # type: ignore - client_with_events.run_task(TaskRequest(name="foo", instrument_session="cm12345-1")) + client_with_events.run_task( + TaskRequestV2(name="foo", instrument_session="cm12345-1") + ) assert failing_callback.mock_calls == [call(evt), call(COMPLETE_EVENT)] assert callback.mock_calls == [call(evt), call(COMPLETE_EVENT)] @@ -1111,7 +1119,7 @@ def test_client_logout(mock_rest: Mock, client: BlueapiClient): @pytest.mark.parametrize("event", [COMPLETE_EVENT, FAILED_EVENT]) def test_run_blocking(event: WorkerEvent, client: BlueapiClient, mock_rest: Mock): mock_rest.run_blocking.side_effect = lambda req: [event] - res = client.run_blocking(TaskRequest(name="foo", instrument_session="cm12345-1")) + res = client.run_blocking(TaskRequestV2(name="foo", instrument_session="cm12345-1")) assert res == event.task_status diff --git a/tests/unit_tests/client/test_rest.py b/tests/unit_tests/client/test_rest.py index 7aa2870886..1ef923e1ec 100644 --- a/tests/unit_tests/client/test_rest.py +++ b/tests/unit_tests/client/test_rest.py @@ -34,17 +34,19 @@ DeviceModel, EnvironmentResponse, PlanModel, - TaskRequest, + TaskRequestV2, TaskResponse, TasksListResponse, WorkerTask, ) from blueapi.worker.event import WorkerState -from blueapi.worker.task import Task, TaskParams +from blueapi.worker.task import Task, TaskParametersV2 from blueapi.worker.task_worker import TrackableTask -TASK_REQUEST = TaskRequest( - name="foo", params=TaskParams(kwargs={"one": "two"}), instrument_session="cm12345-1" +TASK_REQUEST = TaskRequestV2( + name="foo", + params=TaskParametersV2(kwargs={"one": "two"}), + instrument_session="cm12345-1", ) @@ -91,10 +93,10 @@ def test_rest_error_code( def test_create_task_serialization(): rest = Mock(spec=BlueapiRestClient) - request = TaskRequest( + request = TaskRequestV2( name="demo", instrument_session="cm12345-1", - params=TaskParams( + params=TaskParametersV2( kwargs={"devices": [DeviceRef(name="foo", cache=Mock(), model=Mock())]} ), ) @@ -119,10 +121,10 @@ class CustomType: pass rest = Mock(spec=BlueapiRestClient) - request = TaskRequest( + request = TaskRequestV2( name="demo", instrument_session="cm12345-1", - params=TaskParams(kwargs={"devices": [CustomType()]}), + params=TaskParametersV2(kwargs={"devices": [CustomType()]}), ) with pytest.raises(PydanticSerializationError, match="not serializable"): diff --git a/tests/unit_tests/service/test_authorization.py b/tests/unit_tests/service/test_authorization.py index 2be684c6bd..0fddbde2fc 100644 --- a/tests/unit_tests/service/test_authorization.py +++ b/tests/unit_tests/service/test_authorization.py @@ -13,8 +13,8 @@ submit_permission, validate_tiled_config, ) -from blueapi.service.model import TaskRequest -from blueapi.worker import TaskParams +from blueapi.service.model import TaskRequestV2 +from blueapi.worker import TaskParametersV2 # Reusable client patch decorator patch_client_session = patch( @@ -195,7 +195,9 @@ async def test_user_client_can_submit_task(result, context: AbstractContextManag with context: await user_client.can_submit_task( - TaskRequest(name="foo", params=TaskParams(), instrument_session="cm12345-1") + TaskRequestV2( + name="foo", params=TaskParametersV2(), instrument_session="cm12345-1" + ) ) opa.require_submit_task.assert_called_once_with("cm12345-1", "foo_bar") diff --git a/tests/unit_tests/service/test_interface.py b/tests/unit_tests/service/test_interface.py index 94617381c0..5bd163c7dd 100644 --- a/tests/unit_tests/service/test_interface.py +++ b/tests/unit_tests/service/test_interface.py @@ -34,7 +34,8 @@ ProtocolInfo, PythonEnvironmentResponse, SourceInfo, - TaskRequest, + TaskRequestV1, + TaskRequestV2, WorkerTask, ) from blueapi.utils.invalid_config_error import InvalidConfigError @@ -46,7 +47,7 @@ WorkerEvent, WorkerState, ) -from blueapi.worker.task import Task, TaskParams +from blueapi.worker.task import Task, TaskParametersV2 from blueapi.worker.task_worker import TrackableTask FAKE_INSTRUMENT_SESSION = "cm12345-1" @@ -192,10 +193,26 @@ def test_get_device(context_mock: MagicMock): @patch("blueapi.service.interface.context") -def test_submit_task(context_mock: MagicMock): +def test_submit_task_v2(context_mock: MagicMock): context = BlueskyContext() context.register_plan(my_plan) - task = TaskRequest( + task = TaskRequestV2( + name="my_plan", + instrument_session=FAKE_INSTRUMENT_SESSION, + ) + context_mock.return_value = context + mock_uuid_value = "8dfbb9c2-7a15-47b6-bea8-b6b77c31d3d9" + with patch.object(uuid, "uuid4") as uuid_mock: + uuid_mock.return_value = uuid.UUID(mock_uuid_value) + task_uuid = interface.submit_task(task) + assert task_uuid == mock_uuid_value + + +@patch("blueapi.service.interface.context") +def test_submit_task_v1_converts_to_v2(context_mock: MagicMock): + context = BlueskyContext() + context.register_plan(my_plan) + task = TaskRequestV1( name="my_plan", instrument_session=FAKE_INSTRUMENT_SESSION, ) @@ -211,7 +228,7 @@ def test_submit_task(context_mock: MagicMock): def test_clear_task(context_mock: MagicMock): context = BlueskyContext() context.register_plan(my_plan) - task = TaskRequest( + task = TaskRequestV2( name="my_plan", instrument_session=FAKE_INSTRUMENT_SESSION, ) @@ -361,7 +378,7 @@ def test_get_task_by_id( context_mock.return_value = context task_id = interface.submit_task( - TaskRequest( + TaskRequestV2( name="my_plan", instrument_session=FAKE_INSTRUMENT_SESSION, ), @@ -385,7 +402,7 @@ def test_get_task_by_id( request_id=ANY, task=Task( name="my_plan", - params=TaskParams(), + params=TaskParametersV2(), metadata=expected_metadata, ), is_complete=False, @@ -403,7 +420,7 @@ def test_submit_task_inserts_metadata(context_mock: MagicMock): metadata = {"foo": "bar"} task_id = interface.submit_task( - TaskRequest( + TaskRequestV2( name="my_plan", instrument_session=FAKE_INSTRUMENT_SESSION, ), @@ -417,7 +434,7 @@ def test_submit_task_inserts_metadata(context_mock: MagicMock): request_id=ANY, task=Task( name="my_plan", - params=TaskParams(), + params=TaskParametersV2(), metadata=expected_metadata, ), is_complete=False, diff --git a/tests/unit_tests/service/test_protocol.py b/tests/unit_tests/service/test_protocol.py index 08a0e39130..7f758d0cf8 100644 --- a/tests/unit_tests/service/test_protocol.py +++ b/tests/unit_tests/service/test_protocol.py @@ -4,7 +4,7 @@ from pydantic import ValidationError from pydantic_core import InitErrorDetails -from blueapi.service.model import TaskRequest +from blueapi.service.model import TaskRequestV2 from blueapi.service.protocol import ( Abort, ArgumentError, @@ -15,7 +15,7 @@ Resume, Submit, ) -from blueapi.worker import TaskParams +from blueapi.worker import TaskParametersV2 @pytest.mark.parametrize( @@ -30,8 +30,10 @@ } }""", Submit( - task=TaskRequest( - name="foo", params=TaskParams(), instrument_session="cm12345-1" + task=TaskRequestV2( + name="foo", + params=TaskParametersV2(), + instrument_session="cm12345-1", ) ), ), diff --git a/tests/unit_tests/service/test_rest_api.py b/tests/unit_tests/service/test_rest_api.py index 9da3a754cb..5f72645c03 100644 --- a/tests/unit_tests/service/test_rest_api.py +++ b/tests/unit_tests/service/test_rest_api.py @@ -42,12 +42,12 @@ PythonEnvironmentResponse, SourceInfo, StateChangeRequest, - TaskRequest, + TaskRequestV2, WorkerTask, ) from blueapi.service.runner import WorkerDispatcher from blueapi.worker.event import TaskStatus, WorkerEvent, WorkerState -from blueapi.worker.task import Task, TaskParams +from blueapi.worker.task import Task, TaskParametersV2 from blueapi.worker.task_worker import TrackableTask @@ -170,9 +170,9 @@ def test_rest_config_with_cors( client_with_cors: TestClient, mock_runner: Mock, ): - task = TaskRequest( + task = TaskRequestV2( name="my-plan", - params=TaskParams(kwargs={"id": "x"}), + params=TaskParametersV2(kwargs={"id": "x"}), instrument_session=FAKE_INSTRUMENT_SESSION, ) task_id = "f8424be3-203c-494e-b22f-219933b4fa67" @@ -285,9 +285,9 @@ def test_get_non_existent_device_by_name(mock_runner: Mock, client: TestClient) def test_create_task(mock_runner: Mock, client: TestClient) -> None: - task = TaskRequest( + task = TaskRequestV2( name="count", - params=TaskParams(kwargs={"detectors": ["x"]}), + params=TaskParametersV2(kwargs={"detectors": ["x"]}), instrument_session=FAKE_INSTRUMENT_SESSION, ) task_id = str(uuid.uuid4()) @@ -306,9 +306,9 @@ def test_submit_task_requires_permission( mock_opa_client: Mock, access_token: str, ): - task = TaskRequest( + task = TaskRequestV2( name="sleep", - params=TaskParams(kwargs={"time": 2}), + params=TaskParametersV2(kwargs={"time": 2}), instrument_session="cm12345-2", ) client_with_opa.headers["Authorization"] = f"Bearer {access_token}" @@ -325,9 +325,9 @@ def test_create_task_inserts_auth_metadata( mock_runner: Mock, client_with_auth: TestClient, ) -> None: - task = TaskRequest( + task = TaskRequestV2( name="count", - params=TaskParams(kwargs={"detectors": ["x"]}), + params=TaskParametersV2(kwargs={"detectors": ["x"]}), instrument_session=FAKE_INSTRUMENT_SESSION, ) client_with_auth.follow_redirects = False @@ -401,7 +401,7 @@ def test_get_tasks(mock_runner: Mock, client: TestClient) -> None: tasks = [ TrackableTask( task_id="0", - task=Task(name="sleep", params=TaskParams(kwargs={"time": 0.0})), + task=Task(name="sleep", params=TaskParametersV2(kwargs={"time": 0.0})), ), TrackableTask( task_id="1", diff --git a/tests/unit_tests/worker/test_task_worker.py b/tests/unit_tests/worker/test_task_worker.py index b4f2460af5..a832724ee6 100644 --- a/tests/unit_tests/worker/test_task_worker.py +++ b/tests/unit_tests/worker/test_task_worker.py @@ -27,7 +27,7 @@ from blueapi.utils.base_model import BlueapiBaseModel from blueapi.worker import ( Task, - TaskParams, + TaskParametersV2, TaskStatus, TaskWorker, TrackableTask, @@ -38,16 +38,16 @@ ) from blueapi.worker.event import TaskResult, TaskStatusEnum -_SIMPLE_TASK = Task(name="sleep", params=TaskParams(kwargs={"time": 0.0})) -_LONG_TASK = Task(name="sleep", params=TaskParams(kwargs={"time": 1.0})) +_SIMPLE_TASK = Task(name="sleep", params=TaskParametersV2(kwargs={"time": 0.0})) +_LONG_TASK = Task(name="sleep", params=TaskParametersV2(kwargs={"time": 1.0})) _INDEFINITE_TASK = Task( name="set_absolute", - params=TaskParams(kwargs={"movable": "fake_device", "value": 4.0}), + params=TaskParametersV2(kwargs={"movable": "fake_device", "value": 4.0}), ) -_FAILING_TASK = Task(name="failing_plan", params=TaskParams()) +_FAILING_TASK = Task(name="failing_plan", params=TaskParametersV2()) _TASK_WITH_METADATA = Task( name="sleep", - params=TaskParams(kwargs={"time": 0.0}), + params=TaskParametersV2(kwargs={"time": 0.0}), metadata={ "foo": "bar", "baz": 0, @@ -524,7 +524,7 @@ def assert_running_count_plan_produces_ordered_worker_and_data_events( timeout: float = 5.0, ) -> None: default_task = Task( - name="count", params=TaskParams(kwargs={"detectors": ["motor"], "num": 1}) + name="count", params=TaskParametersV2(kwargs={"detectors": ["motor"], "num": 1}) ) task = task or default_task @@ -637,7 +637,9 @@ def test_get_tasks(worker: TaskWorker, status, expected_task_ids): task_id="task1", task=Task( name="set_absolute", - params=TaskParams(kwargs={"movable": "fake_device", "value": 4.0}), + params=TaskParametersV2( + kwargs={"movable": "fake_device", "value": 4.0} + ), ), is_complete=False, is_pending=False, @@ -646,7 +648,9 @@ def test_get_tasks(worker: TaskWorker, status, expected_task_ids): task_id="task2", task=Task( name="set_absolute", - params=TaskParams(kwargs={"movable": "fake_device", "value": 4.0}), + params=TaskParametersV2( + kwargs={"movable": "fake_device", "value": 4.0} + ), ), is_complete=False, is_pending=True, @@ -657,7 +661,9 @@ def test_get_tasks(worker: TaskWorker, status, expected_task_ids): task_id="task3", task=Task( name="set_absolute", - params=TaskParams(kwargs={"movable": "fake_device", "value": 4.0}), + params=TaskParametersV2( + kwargs={"movable": "fake_device", "value": 4.0} + ), ), is_complete=True, is_pending=False, @@ -981,7 +987,7 @@ def positional_plan(value: int) -> MsgGenerator: context.register_plan(positional_plan) - task = Task(name="positional_plan", params=TaskParams(args=[42])) + task = Task(name="positional_plan", params=TaskParametersV2(args=[42])) args, kwargs = task.prepare_params(context) assert args == [42] @@ -996,7 +1002,7 @@ def positional_plan(first: int, second: int) -> MsgGenerator: yield from () context.register_plan(positional_plan) - task = Task(name="positional_plan", params=TaskParams(args=[1, 2])) + task = Task(name="positional_plan", params=TaskParametersV2(args=[1, 2])) args, kwargs = task.prepare_params(context) assert args == [1, 2] @@ -1018,7 +1024,7 @@ def mixed_plan( context.register_plan(mixed_plan) task = Task( name="mixed_plan", - params=TaskParams(args=[1, 2], kwargs={"third": 3}), + params=TaskParametersV2(args=[1, 2], kwargs={"third": 3}), ) args, kwargs = task.prepare_params(context) assert args == [1, 2] @@ -1035,7 +1041,7 @@ def positional_device_plan(device: FakeDevice) -> MsgGenerator: context.register_plan(positional_device_plan) task = Task( - name="positional_device_plan", params=TaskParams(args=[fake_device.name]) + name="positional_device_plan", params=TaskParametersV2(args=[fake_device.name]) ) args, kwargs = task.prepare_params(context) assert args == [fake_device] From 23e09b6bc80ce579014e3405d3ada2236e78e65c Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Mon, 12 Oct 2026 20:53:56 +0000 Subject: [PATCH 11/31] Update task parameter test --- tests/unit_tests/cli/test_cli.py | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/tests/unit_tests/cli/test_cli.py b/tests/unit_tests/cli/test_cli.py index 3d9227ae0d..c5534d0567 100644 --- a/tests/unit_tests/cli/test_cli.py +++ b/tests/unit_tests/cli/test_cli.py @@ -1431,10 +1431,23 @@ def test_config_schema( assert json.loads(result.output) == expected -@pytest.mark.parametrize("value,result", [({}, {}), ("{}", {}), (None, None)]) -def test_task_parameter_type(value, result): +@pytest.mark.parametrize( + "value, expected", + [ + ({}, TaskParametersV2()), + ("{}", TaskParametersV2()), + (None, TaskParametersV2()), + ('{"time": 3}', TaskParametersV2(kwargs={"time": 3})), + ( + '{"args": [], "kwargs": {"time": 3}}', + TaskParametersV2(args=[], kwargs={"time": 3}), + ), + ], + ids=["empty-dict", "empty-json", "none", "v1", "v2"], +) +def test_task_parameter_type(value, expected): t = ParametersType() - assert t.convert(value, None, None) == result + assert t.convert(value, None, None) == expected @pytest.mark.parametrize( From 461272df59d17730e2e50a74a4cf32c910ba54c3 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Mon, 12 Oct 2026 21:21:10 +0000 Subject: [PATCH 12/31] Clean up --- src/blueapi/service/model.py | 9 +++------ src/blueapi/worker/task.py | 4 ++-- 2 files changed, 5 insertions(+), 8 deletions(-) diff --git a/src/blueapi/service/model.py b/src/blueapi/service/model.py index 3f133f808d..3a8a5c936d 100644 --- a/src/blueapi/service/model.py +++ b/src/blueapi/service/model.py @@ -1,5 +1,5 @@ import uuid -from collections.abc import Iterable, Mapping +from collections.abc import Iterable from enum import StrEnum from typing import Annotated, Any @@ -11,7 +11,7 @@ from blueapi.core import BLUESKY_PROTOCOLS, Device, Plan from blueapi.core.context import generic_bounds from blueapi.utils import BlueapiBaseModel -from blueapi.worker import TaskParametersV2, WorkerState +from blueapi.worker import TaskParametersV1, TaskParametersV2, WorkerState from blueapi.worker.task_worker import TaskWorker, TrackableTask _UNKNOWN_NAME = "UNKNOWN" @@ -58,16 +58,13 @@ class TasksListResponse(BlueapiBaseModel): tasks: list[TrackableTask] = Field(description="List of tasks") -TaskParametersV2V1 = Mapping[str, Any] - - class TaskRequestV1(BlueapiBaseModel): """ Request to run a task with related info. Supports **kwargs only. """ name: str = Field(description="Name of plan to run") - params: TaskParametersV2V1 = Field( + params: TaskParametersV1 = Field( description="Values for parameters to plan, if any", default_factory=dict ) instrument_session: str = Field( diff --git a/src/blueapi/worker/task.py b/src/blueapi/worker/task.py index 0284b14c93..e9fd1a4b33 100644 --- a/src/blueapi/worker/task.py +++ b/src/blueapi/worker/task.py @@ -48,7 +48,8 @@ def prepare_params( ctx: Context holding plans and devices Returns: - tuple[list[Any], Mapping[str, Any]]: The prepared parameters for the plan. + tuple[list[Any], Mapping[str, Any]]: The prepared *args and **kwargs + parameters for the plan. """ plan = ctx.plans[self.name] func = ctx.plan_functions[self.name] @@ -88,7 +89,6 @@ def prepare_params( case Parameter.VAR_KEYWORD: kwargs.update(value) - else: # Let the generated Pydantic model provide the default. value = getattr(validated, name) From f882f08460ad2e11450ccbd5811587a4fd00f940 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Tue, 13 Oct 2026 22:15:11 +0000 Subject: [PATCH 13/31] Fix typing --- src/blueapi/client/plan_cache.py | 2 +- src/blueapi/worker/task.py | 4 ++-- tests/unit_tests/client/test_plan_cache.py | 10 +++++----- 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/src/blueapi/client/plan_cache.py b/src/blueapi/client/plan_cache.py index 8da1a18107..aa49dbe06a 100644 --- a/src/blueapi/client/plan_cache.py +++ b/src/blueapi/client/plan_cache.py @@ -95,7 +95,7 @@ def properties(self) -> dict[str, Any]: def required(self) -> list[str]: return self.model.parameter_schema.get("required", []) - def _build_args(self, *args, **kwargs) -> TaskParametersV2: + def _build_args(self, *args: Any, **kwargs: Any) -> TaskParametersV2: """Build a parameter mapping from positional and keyword arguments. Positional arguments are assigned to parameters according to their diff --git a/src/blueapi/worker/task.py b/src/blueapi/worker/task.py index 33a57749d0..563f45f369 100644 --- a/src/blueapi/worker/task.py +++ b/src/blueapi/worker/task.py @@ -1,5 +1,5 @@ import logging -from collections.abc import Mapping +from collections.abc import Mapping, Sequence from inspect import Parameter, signature from typing import Any @@ -15,7 +15,7 @@ class TaskParametersV2(BlueapiBaseModel): - args: list[Any] = Field(default_factory=list) + args: Sequence[Any] = Field(default_factory=list) kwargs: Mapping[str, Any] = Field(default_factory=dict) @classmethod diff --git a/tests/unit_tests/client/test_plan_cache.py b/tests/unit_tests/client/test_plan_cache.py index 82b77fd74e..81fca41462 100644 --- a/tests/unit_tests/client/test_plan_cache.py +++ b/tests/unit_tests/client/test_plan_cache.py @@ -135,23 +135,23 @@ def test_plan_empty_fallback_help_text(client): "plan_args", [ pytest.param( - TaskParametersV2(args=[1], kwargs={}), + TaskParametersV2(args=(1,), kwargs={}), id="required_as_positional", ), pytest.param( - TaskParametersV2(args=[], kwargs={"one": 7}), + TaskParametersV2(args=(), kwargs={"one": 7}), id="required_as_keyword", ), pytest.param( - TaskParametersV2(args=[1], kwargs={"two": 23}), + TaskParametersV2(args=(1,), kwargs={"two": 23}), id="all_as_mixed_args_kwargs", ), p( - TaskParametersV2(args=[1, 2], kwargs={}), + TaskParametersV2(args=(1, 2), kwargs={}), id="all_as_positional", ), p( - TaskParametersV2(args=[], kwargs={"one": 21, "two": 42}), + TaskParametersV2(args=(), kwargs={"one": 21, "two": 42}), id="all_as_keyword", ), ], From a3e46efbfcb95b443c3fd20994a6dcb9c23db23c Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 14 Oct 2026 11:01:10 +0000 Subject: [PATCH 14/31] Update openapi --- docs/reference/openapi.yaml | 37 +++++++++++++++++++++++++------------ src/blueapi/config.py | 2 +- 2 files changed, 26 insertions(+), 13 deletions(-) diff --git a/docs/reference/openapi.yaml b/docs/reference/openapi.yaml index 9747fc0968..17f3169fb9 100644 --- a/docs/reference/openapi.yaml +++ b/docs/reference/openapi.yaml @@ -246,10 +246,8 @@ components: title: Name type: string params: - additionalProperties: true + $ref: '#/components/schemas/TaskParametersV2' description: Values for parameters to plan, if any - title: Params - type: object required: - name title: Task @@ -274,9 +272,22 @@ components: - message title: TaskError type: object + TaskParametersV2: + additionalProperties: false + properties: + args: + items: {} + title: Args + type: array + kwargs: + additionalProperties: true + title: Kwargs + type: object + title: TaskParametersV2 + type: object TaskRequestV2: additionalProperties: false - description: Request to run a task with related info + description: Request to run a task with related info. Supports *args and **kwargs. properties: instrument_session: description: Instrument session associated with this task @@ -287,10 +298,8 @@ components: title: Name type: string params: - additionalProperties: true + $ref: '#/components/schemas/TaskParametersV2' description: Values for parameters to plan, if any - title: Params - type: object required: - name - instrument_session @@ -449,7 +458,7 @@ info: name: Apache 2.0 url: https://www.apache.org/licenses/LICENSE-2.0.html title: BlueAPI Control - version: 1.5.0 + version: 1.6.0 openapi: 3.1.0 paths: /api/v1/devices: @@ -646,8 +655,10 @@ paths: - instrument_session: cm12345-1 name: count params: - detectors: - - x + args: [] + kwargs: + detectors: + - x required: true responses: '201': @@ -1047,8 +1058,10 @@ paths: - instrument_session: cm12345-1 name: count params: - detectors: - - x + args: [] + kwargs: + detectors: + - x required: true responses: '201': diff --git a/src/blueapi/config.py b/src/blueapi/config.py index 9c339c71f5..f65dd65f0a 100644 --- a/src/blueapi/config.py +++ b/src/blueapi/config.py @@ -335,7 +335,7 @@ class ApplicationConfig(BlueapiBaseModel): """ #: API version to publish in OpenAPI schema - REST_API_VERSION: ClassVar[str] = "1.5.0" + REST_API_VERSION: ClassVar[str] = "1.6.0" LICENSE_INFO: ClassVar[dict[str, str]] = { "name": "Apache 2.0", From 88a1b48fa7c52426fd7e65b3a8bfcdeb73a223dd Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 14 Oct 2026 13:23:35 +0000 Subject: [PATCH 15/31] Fix lint --- tests/unit_tests/client/constants.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/unit_tests/client/constants.py b/tests/unit_tests/client/constants.py index fad6057c6f..727cd15dc4 100644 --- a/tests/unit_tests/client/constants.py +++ b/tests/unit_tests/client/constants.py @@ -11,6 +11,7 @@ ) from blueapi.worker import Task, TrackableTask, WorkerEvent, WorkerState from blueapi.worker.event import TaskError, TaskResult, TaskStatus +from blueapi.worker.task import TaskParametersV2 PLANS = PlanResponse( plans=[ @@ -39,7 +40,7 @@ ] ) DEVICE = DeviceModel(name="foo", protocols=[]) -TASK = TrackableTask(task_id="foo", task=Task(name="bar", params={})) +TASK = TrackableTask(task_id="foo", task=Task(name="bar", params=TaskParametersV2())) TASKS = TasksListResponse(tasks=[TASK]) ACTIVE_TASK = WorkerTask(task_id="bar") ENVIRONMENT_ID = uuid.uuid4() From bcbd630833fc0bd634713dfc9a3ad509484703e3 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 14 Oct 2026 13:56:41 +0000 Subject: [PATCH 16/31] Add rest tak request v1 and v2 --- tests/unit_tests/client/test_rest.py | 38 +++++++++++++++++++++------- 1 file changed, 29 insertions(+), 9 deletions(-) diff --git a/tests/unit_tests/client/test_rest.py b/tests/unit_tests/client/test_rest.py index 9c0be65218..80348aef75 100644 --- a/tests/unit_tests/client/test_rest.py +++ b/tests/unit_tests/client/test_rest.py @@ -34,6 +34,8 @@ DeviceModel, EnvironmentResponse, PlanModel, + TaskRequest, + TaskRequestV1, TaskRequestV2, TaskResponse, TasksListResponse, @@ -91,17 +93,35 @@ def test_rest_error_code( rest.get_plans() -def test_create_task_serialization(): - rest = Mock(spec=BlueapiRestClient) - request = TaskRequestV2( - name="demo", - instrument_session="cm12345-1", - params=TaskParametersV2( - kwargs={"devices": [DeviceRef(name="foo", cache=Mock(), model=Mock())]} +@pytest.mark.parametrize( + ("task_request", "expected_data"), + [ + pytest.param( + TaskRequestV1( + name="demo", + instrument_session="cm12345-1", + params={"devices": [DeviceRef(name="foo", cache=Mock(), model=Mock())]}, + ), + id="v1", ), - ) + pytest.param( + TaskRequestV2( + name="demo", + instrument_session="cm12345-1", + params=TaskParametersV2( + kwargs={ + "devices": [DeviceRef(name="foo", cache=Mock(), model=Mock())] + } + ), + ), + id="v2", + ), + ], +) +def test_create_task_serialization(task_request: TaskRequest): + rest = Mock(spec=BlueapiRestClient) - BlueapiRestClient.create_task(rest, request) + BlueapiRestClient.create_task(rest, task_request) rest._request_and_deserialize.assert_called_once_with( "/tasks", From 2e9d6f185f45f3486daa391a1b168bf32159d3b6 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 14 Oct 2026 14:07:33 +0000 Subject: [PATCH 17/31] Fix test --- tests/unit_tests/client/test_rest.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit_tests/client/test_rest.py b/tests/unit_tests/client/test_rest.py index 80348aef75..ea19ff655c 100644 --- a/tests/unit_tests/client/test_rest.py +++ b/tests/unit_tests/client/test_rest.py @@ -94,7 +94,7 @@ def test_rest_error_code( @pytest.mark.parametrize( - ("task_request", "expected_data"), + "task_request", [ pytest.param( TaskRequestV1( From 6484e381c96b0cbe93a8f20438e96eba7dcaf8ac Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 14 Oct 2026 14:26:36 +0000 Subject: [PATCH 18/31] update all rest tests to use v1 and v2 --- tests/unit_tests/client/test_rest.py | 41 +++++++++++++++++++++------- 1 file changed, 31 insertions(+), 10 deletions(-) diff --git a/tests/unit_tests/client/test_rest.py b/tests/unit_tests/client/test_rest.py index ea19ff655c..2fedd1ed93 100644 --- a/tests/unit_tests/client/test_rest.py +++ b/tests/unit_tests/client/test_rest.py @@ -45,7 +45,12 @@ from blueapi.worker.task import Task, TaskParametersV2 from blueapi.worker.task_worker import TrackableTask -TASK_REQUEST = TaskRequestV2( +TASK_REQUEST_V1 = TaskRequestV1( + name="foo", + params={"one": "two"}, + instrument_session="cm12345-1", +) +TASK_REQUEST_V2 = TaskRequestV2( name="foo", params=TaskParametersV2(kwargs={"one": "two"}), instrument_session="cm12345-1", @@ -466,14 +471,17 @@ def test_get_missing_plan(rest: BlueapiRestClient): rest.get_plan("foo") +@pytest.mark.parametrize("task_request", [TASK_REQUEST_V1, TASK_REQUEST_V2]) @patch("blueapi.client.rest.connect") -def test_run_blocking(mock_connect: Mock, rest: BlueapiRestClient): +def test_run_blocking( + mock_connect: Mock, rest: BlueapiRestClient, task_request: TaskRequest +): ws = MagicMock() ws.__enter__.return_value.__iter__.return_value = iter( ['{"kind": "update", "data": {"name": "start", "doc":{}, "task_id":"t_uid"}}'] ) mock_connect.return_value = ws - conn = rest.run_blocking(TASK_REQUEST) + conn = rest.run_blocking(task_request) next(iter(conn)) mock_connect.assert_called_once_with( "ws://localhost:8000/api/v2/run_plan", @@ -482,6 +490,7 @@ def test_run_blocking(mock_connect: Mock, rest: BlueapiRestClient): ) +@pytest.mark.parametrize("task_request", [TASK_REQUEST_V1, TASK_REQUEST_V2]) @patch("blueapi.client.rest.connect") def test_run_blocking_auth( mock_connect: Mock, @@ -489,13 +498,14 @@ def test_run_blocking_auth( mock_authn_server: responses.RequestsMock, cached_valid_token: Path, # creates the cache file cached_valid_token_value: str, # the value from the file + task_request: TaskRequest, ): ws = MagicMock() ws.__enter__.return_value.__iter__.return_value = iter( ['{"kind": "update", "data": {"name": "start", "doc":{}, "task_id":"t_uid"}}'] ) mock_connect.return_value = ws - conn = rest_with_auth.run_blocking(TASK_REQUEST) + conn = rest_with_auth.run_blocking(task_request) next(iter(conn)) mock_connect.assert_called_once_with( "ws://localhost:8000/api/v2/run_plan", @@ -504,6 +514,7 @@ def test_run_blocking_auth( ) +@pytest.mark.parametrize("task_request", [TASK_REQUEST_V1, TASK_REQUEST_V2]) @pytest.mark.parametrize( "event,error,message", [ @@ -531,11 +542,12 @@ def test_run_blocking_errors( event: str, error: type[Exception], message: str, + task_request: TaskRequest, ): ws = MagicMock() ws.__enter__.return_value.__iter__.return_value = iter([event]) mock_connect.return_value = ws - conn = rest.run_blocking(TASK_REQUEST) + conn = rest.run_blocking(task_request) with pytest.raises(error, match=message): next(iter(conn)) mock_connect.assert_called_once_with( @@ -545,23 +557,30 @@ def test_run_blocking_errors( ) +@pytest.mark.parametrize("task_request", [TASK_REQUEST_V1, TASK_REQUEST_V2]) @pytest.mark.parametrize("status_code", [401, 403]) @patch("blueapi.client.rest.connect") def test_run_blocking_ws_failures( - mock_connect: Mock, rest: BlueapiRestClient, status_code: int + mock_connect: Mock, + rest: BlueapiRestClient, + status_code: int, + task_request: TaskRequest, ): mock_connect.side_effect = InvalidStatus( response=Response( status_code=status_code, reason_phrase="test_error", headers=Headers() ) ) - conn = rest.run_blocking(TASK_REQUEST) + conn = rest.run_blocking(task_request) with pytest.raises(UnauthorisedAccessError): next(iter(conn)) +@pytest.mark.parametrize("task_request", [TASK_REQUEST_V1, TASK_REQUEST_V2]) @patch("blueapi.client.rest.connect") -def test_run_blocking_unauthorised(mock_connect: Mock, rest: BlueapiRestClient): +def test_run_blocking_unauthorised( + mock_connect: Mock, rest: BlueapiRestClient, task_request: TaskRequest +): ws = MagicMock() event = """{ "kind":"unauthorized" @@ -569,11 +588,12 @@ def test_run_blocking_unauthorised(mock_connect: Mock, rest: BlueapiRestClient): ws.__enter__.return_value.__iter__.return_value = iter([event]) mock_connect.return_value = ws - conn = rest.run_blocking(TASK_REQUEST) + conn = rest.run_blocking(task_request) with pytest.raises(UnauthorisedAccessError): next(iter(conn)) +@pytest.mark.parametrize("task_request", [TASK_REQUEST_V1, TASK_REQUEST_V2]) @pytest.mark.parametrize( "conn_err,exp_err", [ @@ -597,8 +617,9 @@ def test_run_blocking_connect_error( rest: BlueapiRestClient, conn_err: Exception, exp_err: type[Exception], + task_request: TaskRequest, ): mock_connect.side_effect = conn_err - conn = rest.run_blocking(TASK_REQUEST) + conn = rest.run_blocking(task_request) with pytest.raises(exp_err): next(iter(conn)) From b430aa9ba7b636dc932276372e15c0aa55e140b3 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 14 Oct 2026 15:38:14 +0000 Subject: [PATCH 19/31] Add additional tests --- src/blueapi/cli/cli.py | 5 +- src/blueapi/core/context.py | 7 ++ tests/unit_tests/worker/test_task_worker.py | 102 ++++++++++++++++++++ 3 files changed, 110 insertions(+), 4 deletions(-) diff --git a/src/blueapi/cli/cli.py b/src/blueapi/cli/cli.py index 9f4deb4d21..db82d53b71 100644 --- a/src/blueapi/cli/cli.py +++ b/src/blueapi/cli/cli.py @@ -85,10 +85,7 @@ def convert( # V2 parameters have the shape {"args": [...], "kwargs": {...}}. if value and set(value).issubset({"args", "kwargs"}): - try: - return TaskParametersV2.model_validate(value) - except ValueError as exc: - self.fail(f"Invalid V2 parameters: {exc}", param, ctx) + return TaskParametersV2.model_validate(value) # Otherwise, interpret the object as V1 keyword arguments. return TaskParametersV2.from_v1(value) diff --git a/src/blueapi/core/context.py b/src/blueapi/core/context.py index 76ce74e647..d6b95658aa 100644 --- a/src/blueapi/core/context.py +++ b/src/blueapi/core/context.py @@ -462,6 +462,13 @@ def _type_spec_for_function( info = FieldInfo(default_factory=default_factory) else: _type = self._convert_type(arg_type, no_default) + + # Support *args + if para.kind is Parameter.VAR_POSITIONAL: + _type = tuple[_type, ...] + # Support **kwargs + elif para.kind is Parameter.VAR_KEYWORD: + _type = dict[str, _type] match para.default: case Parameter.empty: info = FieldInfo(default_factory=None) diff --git a/tests/unit_tests/worker/test_task_worker.py b/tests/unit_tests/worker/test_task_worker.py index def6fcdbb7..fa7cb8fd2e 100644 --- a/tests/unit_tests/worker/test_task_worker.py +++ b/tests/unit_tests/worker/test_task_worker.py @@ -1033,3 +1033,105 @@ def positional_device_plan(device: FakeDevice) -> MsgGenerator: args, kwargs = task.prepare_params(context) assert args == [fake_device] assert kwargs == {} + + +def test_positional_only_plan(fake_device: FakeDevice) -> None: + def positional_only_plan(val: int, /, device: FakeDevice) -> MsgGenerator: + yield from () + + context = BlueskyContext() + context.register_device(fake_device) + context.register_plan(positional_only_plan) + + task = Task( + name="positional_only_plan", + params=TaskParametersV2( + args=[1], + kwargs={"device": fake_device.name}, + ), + ) + + args, kwargs = task.prepare_params(context) + assert args == [1] + assert kwargs == {"device": fake_device} + + +def test_var_positional_plan(fake_device: FakeDevice) -> None: + def positional_device_plan(val: int, *device: FakeDevice) -> MsgGenerator: + yield from () + + context = BlueskyContext() + context.register_device(fake_device) + context.register_plan(positional_device_plan) + + task = Task( + name="positional_device_plan", + params=TaskParametersV2(args=[1, fake_device.name, fake_device.name]), + ) + args, kwargs = task.prepare_params(context) + assert args == [1, fake_device, fake_device] + assert kwargs == {} + + +def test_positional_only_and_var_positional_plan( + fake_device: FakeDevice, +) -> None: + def positional_device_plan(val: int, /, *device: FakeDevice) -> MsgGenerator: + yield from () + + context = BlueskyContext() + context.register_device(fake_device) + context.register_plan(positional_device_plan) + + task = Task( + name="positional_device_plan", + params=TaskParametersV2( + args=[1, fake_device.name, fake_device.name], + ), + ) + + args, kwargs = task.prepare_params(context) + assert args == [1, fake_device, fake_device] + assert kwargs == {} + + +def test_var_keyword_plan(fake_device: FakeDevice) -> None: + def keyword_device_plan(val: int, **devices: FakeDevice) -> MsgGenerator: + yield from () + + context = BlueskyContext() + context.register_device(fake_device) + context.register_plan(keyword_device_plan) + + task = Task( + name="keyword_device_plan", + params=TaskParametersV2( + args=[1], + kwargs={"first": fake_device.name, "second": fake_device.name}, + ), + ) + args, kwargs = task.prepare_params(context) + assert args == [1] + assert kwargs == {"first": fake_device, "second": fake_device} + + +def test_positional_only_and_var_keyword_plan( + fake_device: FakeDevice, +) -> None: + def positional_only_plan(val: int, /, **devices: FakeDevice) -> MsgGenerator: + yield from () + + context = BlueskyContext() + context.register_device(fake_device) + context.register_plan(positional_only_plan) + + task = Task( + name="positional_only_plan", + params=TaskParametersV2( + args=[1], + kwargs={"first": fake_device.name, "second": fake_device.name}, + ), + ) + args, kwargs = task.prepare_params(context) + assert args == [1] + assert kwargs == {"first": fake_device, "second": fake_device} From 4fffc338674556fce626f16c11228bf77fd115c8 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Wed, 14 Oct 2026 16:06:46 +0000 Subject: [PATCH 20/31] Minor corrections --- src/blueapi/cli/cli.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/src/blueapi/cli/cli.py b/src/blueapi/cli/cli.py index db82d53b71..49f8828aed 100644 --- a/src/blueapi/cli/cli.py +++ b/src/blueapi/cli/cli.py @@ -4,7 +4,7 @@ import stat import sys import textwrap -from collections.abc import Callable +from collections.abc import Callable, Mapping from functools import wraps from pathlib import Path from pprint import pprint @@ -55,13 +55,13 @@ class ParametersType(ParamType): - """CLI input parameter accepting V1 or V2 JSON parameters.""" + """CLI input parameter to accept a JSON object as an argument""" - name = "parameters" + name = "TaskParameters" def convert( self, - value: str | dict[str, Any] | None, + value: str | Mapping[str, Any] | None, param: Parameter | None, ctx: Context | None, ) -> TaskParametersV2: @@ -82,7 +82,6 @@ def convert( param, ctx, ) - # V2 parameters have the shape {"args": [...], "kwargs": {...}}. if value and set(value).issubset({"args", "kwargs"}): return TaskParametersV2.model_validate(value) From 9f214f161abb4fb5c4c2c8d3e5a344f28e0f1be6 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Thu, 15 Oct 2026 08:06:26 +0000 Subject: [PATCH 21/31] Update _build_args doc string --- src/blueapi/client/plan_cache.py | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/src/blueapi/client/plan_cache.py b/src/blueapi/client/plan_cache.py index aa49dbe06a..210cf873bc 100644 --- a/src/blueapi/client/plan_cache.py +++ b/src/blueapi/client/plan_cache.py @@ -61,9 +61,6 @@ def __init__(self, name, model: PlanModel, client: ClientProtocol): def __call__(self, *args, **kwargs) -> Any: """Execute the plan with the supplied arguments. - Positional arguments are mapped to parameters in the order defined by - the plan's parameter schema. Keyword arguments are passed by name. - Returns: The result returned by the plan. @@ -96,11 +93,10 @@ def required(self) -> list[str]: return self.model.parameter_schema.get("required", []) def _build_args(self, *args: Any, **kwargs: Any) -> TaskParametersV2: - """Build a parameter mapping from positional and keyword arguments. + """Check parameters against schema and build TaskParametersV2. - Positional arguments are assigned to parameters according to their - order in the plan's parameter schema. Keyword arguments are then - added by name. + Returns: + TaskParametersV2 Raises: TypeError: If too many positional arguments, unexpected keyword From 8f5fa121be6874d41ce6a6dedb6028c95f6e00c9 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Thu, 15 Oct 2026 10:43:35 +0000 Subject: [PATCH 22/31] Improve _type_spec_for_function logic --- src/blueapi/core/context.py | 28 +++++++++++++++++++--------- 1 file changed, 19 insertions(+), 9 deletions(-) diff --git a/src/blueapi/core/context.py b/src/blueapi/core/context.py index d6b95658aa..c59a70b674 100644 --- a/src/blueapi/core/context.py +++ b/src/blueapi/core/context.py @@ -439,19 +439,30 @@ def _type_spec_for_function( Returns: Mapping of {name: (type, default)} to be used by pydantic for deserialising - function arguments + function arguments """ args = signature(func).parameters types = get_type_hints(func) new_args: dict[str, tuple[type, FieldInfo]] = {} + for name, para in args.items(): arg_type = types.get(name, Parameter.empty) + if arg_type is Parameter.empty: raise ValueError( f"Type annotation is required for '{name}' in '{func.__name__}'" ) - no_default = para.default is Parameter.empty + match para.kind: + case Parameter.VAR_POSITIONAL: + arg_type = tuple[arg_type, ...] + case Parameter.VAR_KEYWORD: + arg_type = dict[str, arg_type] + + no_default = para.default is Parameter.empty and para.kind not in ( + Parameter.VAR_POSITIONAL, + Parameter.VAR_KEYWORD, + ) if ( isclass(arg_type) and (issubclass(arg_type, BaseModel) or is_dataclass(arg_type)) @@ -463,15 +474,14 @@ def _type_spec_for_function( else: _type = self._convert_type(arg_type, no_default) - # Support *args - if para.kind is Parameter.VAR_POSITIONAL: - _type = tuple[_type, ...] - # Support **kwargs - elif para.kind is Parameter.VAR_KEYWORD: - _type = dict[str, _type] match para.default: case Parameter.empty: - info = FieldInfo(default_factory=None) + if para.kind is Parameter.VAR_POSITIONAL: + info = FieldInfo(default_factory=tuple) + elif para.kind is Parameter.VAR_KEYWORD: + info = FieldInfo(default_factory=dict) + else: + info = FieldInfo(default_factory=None) case None: info = FieldInfo(default_factory=lambda: None) case _: From 970f95c93f77a9e694309ae77350208980b7a64f Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Fri, 16 Oct 2026 10:51:49 +0000 Subject: [PATCH 23/31] Improve test to use args and kwargs --- tests/unit_tests/cli/test_cli.py | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/tests/unit_tests/cli/test_cli.py b/tests/unit_tests/cli/test_cli.py index c5534d0567..aa5f2c2c69 100644 --- a/tests/unit_tests/cli/test_cli.py +++ b/tests/unit_tests/cli/test_cli.py @@ -262,16 +262,17 @@ def test_submit_plan_without_stomp(runner: CliRunner): @pytest.mark.parametrize( - "params", + "params, expected", [ - {"time": 3}, - {"args": [], "kwargs": {"time": 3}}, + [{"time": 3}, {"args": [], "kwargs": {"time": 3}}], + [{"args": [], "kwargs": {"time": 3}}, {"args": [], "kwargs": {"time": 3}}], + [{"args": [3], "kwargs": {}}, {"args": [3], "kwargs": {}}], ], - ids=["v1", "v2"], + ids=["v1", "v2_kwargs", "v2_args"], ) @patch("blueapi.client.client.StompClient") @responses.activate -def test_run_plan(stomp_client: StompClient, runner: CliRunner, params): +def test_run_plan(stomp_client: StompClient, runner: CliRunner, params, expected): task_id = "abcd-1234" submit_response = responses.post( url="http://a.fake.host:12345/tasks", @@ -279,7 +280,7 @@ def test_run_plan(stomp_client: StompClient, runner: CliRunner, params): matchers.json_params_matcher( { "name": "sleep", - "params": {"args": [], "kwargs": {"time": 3}}, + "params": expected, "instrument_session": "cm12345-1", } ) @@ -1442,8 +1443,12 @@ def test_config_schema( '{"args": [], "kwargs": {"time": 3}}', TaskParametersV2(args=[], kwargs={"time": 3}), ), + ( + '{"args": [3], "kwargs": {}}', + TaskParametersV2(args=[3], kwargs={}), + ), ], - ids=["empty-dict", "empty-json", "none", "v1", "v2"], + ids=["empty-dict", "empty-json", "none", "v1", "v2_kwargs", "v2_args"], ) def test_task_parameter_type(value, expected): t = ParametersType() From 05b4cfd9519da539430e7a1cc38a18b75ccef426 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Fri, 16 Oct 2026 13:00:17 +0000 Subject: [PATCH 24/31] Update tests for protocol and rest_api to be stronger, cover v1 and v2 --- src/blueapi/service/model.py | 2 +- src/blueapi/service/protocol.py | 4 +- tests/unit_tests/service/test_protocol.py | 29 +++++++++-- tests/unit_tests/service/test_rest_api.py | 59 ++++++++++++++++++----- 4 files changed, 75 insertions(+), 19 deletions(-) diff --git a/src/blueapi/service/model.py b/src/blueapi/service/model.py index 3a8a5c936d..b29b943ade 100644 --- a/src/blueapi/service/model.py +++ b/src/blueapi/service/model.py @@ -95,7 +95,7 @@ def from_v1(cls, task_v1: TaskRequestV1) -> "TaskRequestV2": ) -TaskRequest = TaskRequestV1 | TaskRequestV2 +TaskRequest = TaskRequestV2 | TaskRequestV1 class DeviceRequest(BlueapiBaseModel): diff --git a/src/blueapi/service/protocol.py b/src/blueapi/service/protocol.py index 7c20fa4fbe..8af5819133 100644 --- a/src/blueapi/service/protocol.py +++ b/src/blueapi/service/protocol.py @@ -20,7 +20,7 @@ from pydantic import BaseModel, Field, TypeAdapter, ValidationError from blueapi.core.bluesky_types import DataEvent -from blueapi.service.model import TaskRequestV2 +from blueapi.service.model import TaskRequest from blueapi.worker.event import ProgressEvent, WorkerEvent UNKNOWN_PLAN = 4001 @@ -37,7 +37,7 @@ class ArgumentError(BaseModel): class Submit(BaseModel): kind: Literal["submit"] = "submit" - task: TaskRequestV2 + task: TaskRequest class Pause(BaseModel): diff --git a/tests/unit_tests/service/test_protocol.py b/tests/unit_tests/service/test_protocol.py index 7f758d0cf8..3acf6a47c9 100644 --- a/tests/unit_tests/service/test_protocol.py +++ b/tests/unit_tests/service/test_protocol.py @@ -4,7 +4,7 @@ from pydantic import ValidationError from pydantic_core import InitErrorDetails -from blueapi.service.model import TaskRequestV2 +from blueapi.service.model import TaskRequestV1, TaskRequestV2 from blueapi.service.protocol import ( Abort, ArgumentError, @@ -21,7 +21,7 @@ @pytest.mark.parametrize( "src,res", [ - ( + pytest.param( """{ "kind": "submit", "task": { @@ -36,10 +36,29 @@ instrument_session="cm12345-1", ) ), + id="submit-v2", + ), + pytest.param( + """{ + "kind": "submit", + "task": { + "name": "foo", + "params": {"one": "two"}, + "instrument_session": "cm12345-1" + } + }""", + Submit( + task=TaskRequestV1( + name="foo", + params={"one": "two"}, + instrument_session="cm12345-1", + ) + ), + id="submit-v1", ), - ('{"kind": "pause"}', Pause()), - ('{"kind": "resume"}', Resume()), - ('{"kind": "abort"}', Abort()), + pytest.param('{"kind": "pause"}', Pause(), id="pause"), + pytest.param('{"kind": "resume"}', Resume(), id="resume"), + pytest.param('{"kind": "abort"}', Abort(), id="abort"), ], ) def test_request_deserialization(src: str, res: Any): diff --git a/tests/unit_tests/service/test_rest_api.py b/tests/unit_tests/service/test_rest_api.py index 5f72645c03..8bbbc9497e 100644 --- a/tests/unit_tests/service/test_rest_api.py +++ b/tests/unit_tests/service/test_rest_api.py @@ -58,7 +58,25 @@ class MockCountModel(BaseModel): ... FAKE_INSTRUMENT_SESSION = "cm12345-1" -SUBMIT_REQUEST = { +SUBMIT_REQUEST_V1 = { + "kind": "submit", + "task": { + "name": "foo", + "params": {"one": "two"}, + "instrument_session": "cm12345-1", + }, +} + +SUBMIT_REQUEST_V2_ARGS = { + "kind": "submit", + "task": { + "name": "foo", + "params": {"args": ["two"], "kwargs": {}}, + "instrument_session": "cm12345-1", + }, +} + +SUBMIT_REQUEST_V2_KWARGS = { "kind": "submit", "task": { "name": "foo", @@ -68,6 +86,17 @@ class MockCountModel(BaseModel): ... } +@pytest.fixture( + params=[ + pytest.param(SUBMIT_REQUEST_V1, id="v1"), + pytest.param(SUBMIT_REQUEST_V2_ARGS, id="v2_args"), + pytest.param(SUBMIT_REQUEST_V2_KWARGS, id="v2_kwargs"), + ] +) +def submit_request(request: pytest.FixtureRequest): + return request.param + + @pytest.fixture def mock_runner() -> Mock: return MagicMock(spec=WorkerDispatcher) @@ -1017,7 +1046,9 @@ def test_logout_when_oidc_config_invalid( assert response.status_code == status.HTTP_205_RESET_CONTENT -async def test_websocket_run_plan(mock_runner: Mock, client: TestClient): +async def test_websocket_run_plan( + mock_runner: Mock, client: TestClient, submit_request +): mock_runner.run.side_effect = lambda req, *a, **kw: { interface.get_active_task: None, interface.submit_task: "task_id", @@ -1047,7 +1078,7 @@ async def test_websocket_run_plan(mock_runner: Mock, client: TestClient): ) with client.websocket_connect("/api/v2/run_plan") as ws: - ws.send_json(SUBMIT_REQUEST) + ws.send_json(submit_request) assert ws.receive_json() == { "kind": "update", "data": { @@ -1120,10 +1151,11 @@ def test_websocket_run_plan_submit_error( reason, mock_runner: Mock, client: TestClient, + submit_request, ): mock_runner.run.side_effect = exc with client.websocket_connect("/api/v2/run_plan") as ws: - ws.send_json(SUBMIT_REQUEST) + ws.send_json(submit_request) assert ws.receive_json() == err_message with pytest.raises(WebSocketDisconnect) as disco: ws.receive_text() @@ -1131,13 +1163,15 @@ def test_websocket_run_plan_submit_error( assert disco.value.reason == reason -def test_websocket_run_plan_server_busy(mock_runner: Mock, client: TestClient): +def test_websocket_run_plan_server_busy( + mock_runner: Mock, client: TestClient, submit_request +): mock_runner.run.side_effect = [ "task_id", # submit_task Mock(name="active_task", is_complete=False), ] with client.websocket_connect("/api/v2/run_plan") as ws: - ws.send_json(SUBMIT_REQUEST) + ws.send_json(submit_request) assert ws.receive_json() == {"kind": "busy"} with pytest.raises(WebSocketDisconnect) as disco: ws.receive_text() @@ -1146,7 +1180,9 @@ def test_websocket_run_plan_server_busy(mock_runner: Mock, client: TestClient): pass -def test_websocket_run_plan_unrelated_events(mock_runner: Mock, client: TestClient): +def test_websocket_run_plan_unrelated_events( + mock_runner: Mock, client: TestClient, submit_request +): mock_runner.run.side_effect = lambda req, *a, **kw: { interface.get_active_task: None, interface.submit_task: "task_id", @@ -1176,7 +1212,7 @@ def test_websocket_run_plan_unrelated_events(mock_runner: Mock, client: TestClie ) with client.websocket_connect("/api/v2/run_plan") as ws: - ws.send_json(SUBMIT_REQUEST) + ws.send_json(submit_request) # first event is not sent assert ws.receive_json() == { "kind": "update", @@ -1202,7 +1238,7 @@ def test_websocket_run_plan_unrelated_events(mock_runner: Mock, client: TestClie def test_websocket_run_plan_client_disconnect_cancels( - mock_runner: Mock, client: TestClient + mock_runner: Mock, client: TestClient, submit_request ): mock_runner.run.side_effect = ["task_id", None, None, None] mock_runner.event_pipe.return_value = contextlib.nullcontext( @@ -1234,7 +1270,7 @@ async def local_send(message: Message): cast(FastAPI, client.app).add_middleware(Disconnector) with client.websocket_connect("/api/v2/run_plan") as ws: - ws.send_json(SUBMIT_REQUEST) + ws.send_json(submit_request) mock_runner.run.assert_called_with( interface.cancel_active_task, failure=True, reason="Client disconnected" ) @@ -1256,6 +1292,7 @@ def test_websocket_run_plan_needs_auth_token( def test_websocket_run_plan_needs_permission( mock_runner: Mock, client_with_opa: TestClient, + submit_request, mock_opa_client: Mock, access_token: str, ): @@ -1263,7 +1300,7 @@ def test_websocket_run_plan_needs_permission( mock_opa_client.can_submit_task.side_effect = HTTPException(status_code=403) mock_runner.run.side_effect = RuntimeError("Task should not be submitted") with client_with_opa.websocket_connect("/api/v2/run_plan") as ws: - ws.send_json(SUBMIT_REQUEST) + ws.send_json(submit_request) assert ws.receive_json() == {"kind": "unauthorized"} with pytest.raises(WebSocketDisconnect) as discon: ws.receive_text() From 38fb396fda7f5a489d43ae36f1486145f9432e69 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Fri, 16 Oct 2026 13:21:52 +0000 Subject: [PATCH 25/31] Add more combinations for system tests --- tests/system_tests/test_blueapi_system.py | 144 ++++++++++++++++++++-- 1 file changed, 131 insertions(+), 13 deletions(-) diff --git a/tests/system_tests/test_blueapi_system.py b/tests/system_tests/test_blueapi_system.py index c3b564f55b..8c352aba5c 100644 --- a/tests/system_tests/test_blueapi_system.py +++ b/tests/system_tests/test_blueapi_system.py @@ -33,6 +33,7 @@ DeviceResponse, PlanResponse, TaskParametersV2, + TaskRequestV1, TaskRequestV2, TaskResponse, WorkerTask, @@ -540,17 +541,99 @@ def test_delete_current_environment(client: BlueapiClient): @pytest.mark.parametrize( - "task,user", + "task", [ - ( + pytest.param( + TaskRequestV1( + name="count", + params={"detectors": ["det"], "num": 3}, + instrument_session="", # Set below + ), + id="v1", + ), + pytest.param( + TaskRequestV1( + name="spec_scan", + params={ + "detectors": ["det"], + "spec": { + "outer": { + "axis": "stage.x", + "start": 0.0, + "stop": 0.4, + "num": 2, + "type": "Linspace", + }, + "inner": { + "axis": "stage.theta", + "start": 5.0, + "stop": 5.3, + "num": 3, + "type": "Linspace", + }, + "gap": True, + "type": "Product", + }, + }, + instrument_session="", + ), + id="spec_scan-v1", + ), + pytest.param( + TaskRequestV2( + name="count", + params=TaskParametersV2(args=[["det"], 3]), + instrument_session="", + ), + id="v2_args", + ), + pytest.param( TaskRequestV2( name="count", params=TaskParametersV2(kwargs={"detectors": ["det"], "num": 3}), - instrument_session=VALID_INSTRUMENT_SESSION[User.alice], + instrument_session="", + ), + id="v2_kwargs", + ), + pytest.param( + TaskRequestV2( + name="count", + params=TaskParametersV2(args=["det"], kwargs={"num": 3}), + instrument_session="", + ), + id="v2_args_kwargs", + ), + pytest.param( + TaskRequestV2( + name="spec_scan", + params=TaskParametersV2( + args=["det"], + kwargs={ + "spec": { + "outer": { + "axis": "stage.x", + "start": 0.0, + "stop": 0.4, + "num": 2, + "type": "Linspace", + }, + "inner": { + "axis": "stage.theta", + "start": 5.0, + "stop": 5.3, + "num": 3, + "type": "Linspace", + }, + "gap": True, + "type": "Product", + }, + }, + ), + instrument_session="", ), - User.alice, + id="spec_scan-v2_args_kwargs", ), - ( + pytest.param( TaskRequestV2( name="spec_scan", params=TaskParametersV2( @@ -576,12 +659,13 @@ def test_delete_current_environment(client: BlueapiClient): }, } ), - instrument_session=VALID_INSTRUMENT_SESSION[User.bob], + instrument_session="", ), - User.bob, + id="spec_scan-v2_kwargs", ), ], ) +@pytest.mark.parametrize("user", [User.alice, User.bob]) @pytest.mark.parametrize("run_method", ["run_task", "run_blocking"]) def test_plan_runs( client_with_stomp: BlueapiClient, @@ -590,6 +674,10 @@ def test_plan_runs( user: ValidUser, run_method: str, ): + # Set instrument session after creation so we can test every single task combination + # with every user + task.instrument_session = VALID_INSTRUMENT_SESSION[user] + resource = Queue(maxsize=1) start = Queue(maxsize=1) @@ -641,15 +729,45 @@ def on_event(event: AnyEvent) -> None: @pytest.mark.parametrize( "task", [ - TaskRequestV2( - name="set_absolute", - params=TaskParametersV2( - kwargs={ + pytest.param( + TaskRequestV1( + name="set_absolute", + params={ "movable": "stage.x", "value": 1.0, - } + }, + instrument_session=VALID_INSTRUMENT_SESSION[User.alice], ), - instrument_session=VALID_INSTRUMENT_SESSION[User.alice], + id="v1", + ), + pytest.param( + TaskRequestV2( + name="set_absolute", + params=TaskParametersV2(args=["stage.x", 1.0]), + instrument_session=VALID_INSTRUMENT_SESSION[User.alice], + ), + id="v2_args", + ), + pytest.param( + TaskRequestV2( + name="set_absolute", + params=TaskParametersV2(args=["stage.x"], kwargs={"value": 1.0}), + instrument_session=VALID_INSTRUMENT_SESSION[User.alice], + ), + id="v2_args_kwargs", + ), + pytest.param( + TaskRequestV2( + name="set_absolute", + params=TaskParametersV2( + kwargs={ + "movable": "stage.x", + "value": 1.0, + } + ), + instrument_session=VALID_INSTRUMENT_SESSION[User.alice], + ), + id="v2_kwargs", ), ], ) From 38c4835a7244a85788c4b7628f7bbe8bae9618c0 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Fri, 16 Oct 2026 13:53:41 +0000 Subject: [PATCH 26/31] Fix system tests, discovered blueapi bug where it freezes when plan given malformed arguments for plan --- tests/system_tests/test_blueapi_system.py | 16 ++++------------ 1 file changed, 4 insertions(+), 12 deletions(-) diff --git a/tests/system_tests/test_blueapi_system.py b/tests/system_tests/test_blueapi_system.py index 8c352aba5c..6195173f3d 100644 --- a/tests/system_tests/test_blueapi_system.py +++ b/tests/system_tests/test_blueapi_system.py @@ -598,7 +598,7 @@ def test_delete_current_environment(client: BlueapiClient): pytest.param( TaskRequestV2( name="count", - params=TaskParametersV2(args=["det"], kwargs={"num": 3}), + params=TaskParametersV2(args=[["det"]], kwargs={"num": 3}), instrument_session="", ), id="v2_args_kwargs", @@ -607,7 +607,7 @@ def test_delete_current_environment(client: BlueapiClient): TaskRequestV2( name="spec_scan", params=TaskParametersV2( - args=["det"], + args=[["det"]], kwargs={ "spec": { "outer": { @@ -732,10 +732,7 @@ def on_event(event: AnyEvent) -> None: pytest.param( TaskRequestV1( name="set_absolute", - params={ - "movable": "stage.x", - "value": 1.0, - }, + params={"movable": "stage.x", "value": 1.0}, instrument_session=VALID_INSTRUMENT_SESSION[User.alice], ), id="v1", @@ -759,12 +756,7 @@ def on_event(event: AnyEvent) -> None: pytest.param( TaskRequestV2( name="set_absolute", - params=TaskParametersV2( - kwargs={ - "movable": "stage.x", - "value": 1.0, - } - ), + params=TaskParametersV2(kwargs={"movable": "stage.x", "value": 1.0}), instrument_session=VALID_INSTRUMENT_SESSION[User.alice], ), id="v2_kwargs", From 48f07035e941b722d1588d996fae8c4ec87d1a45 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Fri, 16 Oct 2026 15:28:48 +0000 Subject: [PATCH 27/31] Fix annotation for main submit_task --- src/blueapi/service/main.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/blueapi/service/main.py b/src/blueapi/service/main.py index a65d5e0569..64b01cbd59 100644 --- a/src/blueapi/service/main.py +++ b/src/blueapi/service/main.py @@ -77,6 +77,7 @@ PythonEnvironmentResponse, SourceInfo, StateChangeRequest, + TaskRequest, TaskRequestV2, TaskResponse, TasksListResponse, @@ -337,7 +338,7 @@ def get_device_by_name( def submit_task( request: Request, response: Response, - task_request: Annotated[TaskRequestV2, Body(..., examples=[example_task_request])], + task_request: Annotated[TaskRequest, Body(..., examples=[example_task_request])], _: Annotated[None, Depends(submit_permission)], runner: Annotated[WorkerDispatcher, Depends(_runner)], fedid: Fedid, From ab1a9a45db8b36064d04b05a60ace31122958b07 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Fri, 16 Oct 2026 15:38:33 +0000 Subject: [PATCH 28/31] Update openapi.yaml --- docs/reference/openapi.yaml | 32 ++++++++++++++++++++++++++++++-- 1 file changed, 30 insertions(+), 2 deletions(-) diff --git a/docs/reference/openapi.yaml b/docs/reference/openapi.yaml index 17f3169fb9..d4ccda4bd0 100644 --- a/docs/reference/openapi.yaml +++ b/docs/reference/openapi.yaml @@ -285,6 +285,28 @@ components: type: object title: TaskParametersV2 type: object + TaskRequestV1: + additionalProperties: false + description: Request to run a task with related info. Supports **kwargs only. + properties: + instrument_session: + description: Instrument session associated with this task + title: Instrument Session + type: string + name: + description: Name of plan to run + title: Name + type: string + params: + additionalProperties: true + description: Values for parameters to plan, if any + title: Params + type: object + required: + - name + - instrument_session + title: TaskRequestV1 + type: object TaskRequestV2: additionalProperties: false description: Request to run a task with related info. Supports *args and **kwargs. @@ -650,7 +672,9 @@ paths: content: application/json: schema: - $ref: '#/components/schemas/TaskRequestV2' + anyOf: + - $ref: '#/components/schemas/TaskRequestV2' + - $ref: '#/components/schemas/TaskRequestV1' examples: - instrument_session: cm12345-1 name: count @@ -659,6 +683,7 @@ paths: kwargs: detectors: - x + title: Task Request required: true responses: '201': @@ -1053,7 +1078,9 @@ paths: content: application/json: schema: - $ref: '#/components/schemas/TaskRequestV2' + anyOf: + - $ref: '#/components/schemas/TaskRequestV2' + - $ref: '#/components/schemas/TaskRequestV1' examples: - instrument_session: cm12345-1 name: count @@ -1062,6 +1089,7 @@ paths: kwargs: detectors: - x + title: Task Request required: true responses: '201': From 7e368aaa1a032bdd3d39ccb4225ab22bb9187698 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Fri, 16 Oct 2026 15:45:22 +0000 Subject: [PATCH 29/31] Implement import feedback. --- src/blueapi/cli/cli.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/src/blueapi/cli/cli.py b/src/blueapi/cli/cli.py index 49f8828aed..070aeed736 100644 --- a/src/blueapi/cli/cli.py +++ b/src/blueapi/cli/cli.py @@ -31,7 +31,10 @@ UnauthorisedAccessError, UnknownPlanError, ) -from blueapi.config import ApplicationConfig, ConfigLoader +from blueapi.config import ( + ApplicationConfig, + ConfigLoader, +) from blueapi.core import OTLP_EXPORT_ENABLED, DataEvent from blueapi.log import set_up_logging from blueapi.service.authentication import SessionCacheManager, SessionManager @@ -41,11 +44,7 @@ SourceInfo, TaskRequestV2, ) -from blueapi.worker import ( - ProgressEvent, - TaskParametersV2, - WorkerEvent, -) +from blueapi.worker import ProgressEvent, TaskParametersV2, WorkerEvent from blueapi.worker.event import TaskError, TaskResult from .scratch import setup_scratch From e517e7308590bd788f5ef00de5aa88b767de4f6b Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Fri, 16 Oct 2026 16:09:19 +0000 Subject: [PATCH 30/31] Add feedback, correctly raise ValidationError in do_task and add test --- src/blueapi/worker/task.py | 2 +- tests/unit_tests/worker/test_task_worker.py | 9 +++++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/src/blueapi/worker/task.py b/src/blueapi/worker/task.py index 563f45f369..4677136582 100644 --- a/src/blueapi/worker/task.py +++ b/src/blueapi/worker/task.py @@ -55,7 +55,7 @@ def prepare_params( func = ctx.plan_functions[self.name] sig = signature(func) - bound = sig.bind(*self.params.args, **self.params.kwargs) + bound = sig.bind_partial(*self.params.args, **self.params.kwargs) # Only validate explicitly supplied arguments. This allows Pydantic's # default/default_factory to provide injected defaults. diff --git a/tests/unit_tests/worker/test_task_worker.py b/tests/unit_tests/worker/test_task_worker.py index fa7cb8fd2e..35f1a8e4bc 100644 --- a/tests/unit_tests/worker/test_task_worker.py +++ b/tests/unit_tests/worker/test_task_worker.py @@ -427,6 +427,15 @@ def test_metadata_passed_to_context(context: BlueskyContext): assert metadata in context.run_engine.md.items() +def test_do_task_with_invalid_parameters_raises_validation_error( + context: BlueskyContext, +): + context.run_engine = Mock() + task = Task(name="sleep", params=TaskParametersV2()) + with pytest.raises(pydantic.ValidationError): + task.do_task(context) + + # # Worker helpers # From 5ff57de1127b3839e2fa88af6090ae41d599ce56 Mon Sep 17 00:00:00 2001 From: Oli Wenman Date: Fri, 16 Oct 2026 16:39:13 +0000 Subject: [PATCH 31/31] Add comment why test is needed --- tests/unit_tests/worker/test_task_worker.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/unit_tests/worker/test_task_worker.py b/tests/unit_tests/worker/test_task_worker.py index 35f1a8e4bc..3c982a5bff 100644 --- a/tests/unit_tests/worker/test_task_worker.py +++ b/tests/unit_tests/worker/test_task_worker.py @@ -432,6 +432,8 @@ def test_do_task_with_invalid_parameters_raises_validation_error( ): context.run_engine = Mock() task = Task(name="sleep", params=TaskParametersV2()) + # Required as this is the error that blueapi/service/main:submit_task is expecting + # to catch. with pytest.raises(pydantic.ValidationError): task.do_task(context)