feat!: Remove non-cli logic from cli.run_plan

Argument parsing has been moved into a click option validator, and task
creation has been moved into the client create_task methods.

This is a breaking change in the BlueapiClient as task methods now
accept plan name and parameters as separate arguments.

Author: tpoliaw

src/blueapi/cli/cli.py

@@ -6,19 +6,21 @@
 from functools import wraps
 from pathlib import Path
 from pprint import pprint
+from typing import Any
 
 import click
 from bluesky.callbacks.best_effort import BestEffortCallback
 from bluesky_stomp.messaging import MessageContext, StompClient
 from bluesky_stomp.models import Broker
+from click.core import Context, Parameter
 from click.exceptions import ClickException
+from click.types import ParamType
 from observability_utils.tracing import setup_tracing
-from pydantic import ValidationError
 from requests.exceptions import ConnectionError
 
 from blueapi import __version__, config
 from blueapi.cli.format import OutputFormat
-from blueapi.client.client import BlueapiClient
+from blueapi.client.client import BlueapiClient, TaskParameters
 from blueapi.client.event_bus import AnyEvent, BlueskyStreamingError, EventBusClient
 from blueapi.client.rest import (
     BlueskyRemoteControlError,
@@ -34,12 +36,27 @@
 from blueapi.log import set_up_logging
 from blueapi.service.authentication import SessionCacheManager, SessionManager
 from blueapi.service.model import SourceInfo
-from blueapi.worker import ProgressEvent, Task, WorkerEvent
+from blueapi.worker import ProgressEvent, WorkerEvent
 
 from .scratch import setup_scratch
 from .updates import CliEventRenderer
 
 
+class ParametersType(ParamType):
+    name = "TaskParameters"
+
+    def convert(
+        self, value: Any, param: Parameter | None, ctx: Context | None
+    ) -> TaskParameters:
+        if isinstance(value, str):
+            try:
+                return json.loads(value)
+            except json.JSONDecodeError as jde:
+                self.fail(f"Parameters are not valid JSON: {jde}")
+        else:
+            return super().convert(value, param, ctx)
+
+
 @click.group(invoke_without_command=True)
 @click.version_option(version=__version__, prog_name="blueapi")
 @click.option(
@@ -218,7 +235,7 @@ def on_event(
 
 @controller.command(name="run")
 @click.argument("name", type=str)
-@click.argument("parameters", type=str, required=False)
+@click.argument("parameters", type=ParametersType(), default={}, required=False)
 @click.option(
     "--foreground/--background", "--fg/--bg", type=bool, is_flag=True, default=True
 )
@@ -234,25 +251,13 @@ def on_event(
 def run_plan(
     obj: dict,
     name: str,
-    parameters: str | None,
     timeout: float | None,
     foreground: bool,
+    parameters: TaskParameters,
 ) -> None:
     """Run a plan with parameters"""
     client: BlueapiClient = obj["client"]
 
-    parameters = parameters or "{}"
-    try:
-        parsed_params = json.loads(parameters) if isinstance(parameters, str) else {}
-    except json.JSONDecodeError as jde:
-        raise ClickException(f"Parameters are not valid JSON: {jde}") from jde
-
-    try:
-        task = Task(name=name, params=parsed_params)
-    except ValidationError as ve:
-        ip = InvalidParameters.from_validation_error(ve)
-        raise ClickException(ip.message()) from ip
-
     try:
         if foreground:
             progress_bar = CliEventRenderer()
@@ -264,12 +269,12 @@ def on_event(event: AnyEvent) -> None:
                 elif isinstance(event, DataEvent):
                     callback(event.name, event.doc)
 
-            resp = client.run_task(task, on_event=on_event)
+            resp = client.run_task(name, parameters, on_event=on_event)
 
             if resp.task_status is not None and not resp.task_status.task_failed:
                 print("Plan Succeeded")
         else:
-            server_task = client.create_and_start_task(task)
+            server_task = client.create_and_start_task(name, parameters)
             click.echo(server_task.task_id)
     except config.MissingStompConfiguration as mse:
         raise ClickException(*mse.args) from mse

src/blueapi/client/client.py

@@ -1,12 +1,14 @@
 import time
 from concurrent.futures import Future
+from typing import Any
 
 from bluesky_stomp.messaging import MessageContext, StompClient
 from bluesky_stomp.models import Broker
 from observability_utils.tracing import (
     get_tracer,
     start_as_current_span,
 )
+from pydantic import ValidationError
 
 from blueapi.config import ApplicationConfig, MissingStompConfiguration
 from blueapi.core.bluesky_types import DataEvent
@@ -28,10 +30,12 @@
 from blueapi.worker.event import ProgressEvent, TaskStatus
 
 from .event_bus import AnyEvent, BlueskyStreamingError, EventBusClient, OnAnyEvent
-from .rest import BlueapiRestClient, BlueskyRemoteControlError
+from .rest import BlueapiRestClient, BlueskyRemoteControlError, InvalidParameters
 
 TRACER = get_tracer("client")
 
+TaskParameters = dict[str, Any]
+
 
 class BlueapiClient:
     """Unified client for controlling blueapi"""
@@ -194,10 +198,12 @@ def get_active_task(self) -> WorkerTask:
 
         return self._rest.get_active_task()
 
-    @start_as_current_span(TRACER, "task", "timeout")
+    @start_as_current_span(TRACER, "name", "parameters", "timeout")
     def run_task(
         self,
-        task: Task,
+        name: str,
+        parameters: TaskParameters = {},
+        *,
         on_event: OnAnyEvent | None = None,
         timeout: float | None = None,
     ) -> WorkerEvent:
@@ -220,7 +226,7 @@ def run_task(
                 "Stomp configuration required to run plans is missing or disabled"
             )
 
-        task_response = self.create_task(task)
+        task_response = self.create_task(name, parameters)
         task_id = task_response.task_id
 
         complete: Future[WorkerEvent] = Future()
@@ -257,8 +263,10 @@ def inner_on_event(event: AnyEvent, ctx: MessageContext) -> None:
             self.start_task(WorkerTask(task_id=task_id))
             return complete.result(timeout=timeout)
 
-    @start_as_current_span(TRACER, "task")
-    def create_and_start_task(self, task: Task) -> TaskResponse:
+    @start_as_current_span(TRACER, "name", "parameters")
+    def create_and_start_task(
+        self, name: str, parameters: TaskParameters = {}
+    ) -> TaskResponse:
         """
         Create a new task and instruct the worker to start it
         immediately.
@@ -270,7 +278,7 @@ def create_and_start_task(self, task: Task) -> TaskResponse:
             TaskResponse: Acknowledgement of request
         """
 
-        response = self.create_task(task)
+        response = self.create_task(name, parameters)
         worker_response = self.start_task(WorkerTask(task_id=response.task_id))
         if worker_response.task_id == response.task_id:
             return response
@@ -280,8 +288,8 @@ def create_and_start_task(self, task: Task) -> TaskResponse:
                 f"but {worker_response.task_id} was started instead"
             )
 
-    @start_as_current_span(TRACER, "task")
-    def create_task(self, task: Task) -> TaskResponse:
+    @start_as_current_span(TRACER, "name", "parameters")
+    def create_task(self, name: str, parameters: TaskParameters = {}) -> TaskResponse:
         """
         Create a new task, does not start execution
 
@@ -292,6 +300,10 @@ def create_task(self, task: Task) -> TaskResponse:
             TaskResponse: Acknowledgement of request
         """
 
+        try:
+            task = Task(name=name, params=parameters)
+        except ValidationError as ve:
+            raise InvalidParameters.from_validation_error(ve)
         return self._rest.create_task(task)
 
     @start_as_current_span(TRACER)

tests/unit_tests/client/test_client.py

@@ -171,15 +171,15 @@ def test_create_task(
     client: BlueapiClient,
     mock_rest: Mock,
 ):
-    client.create_task(task=Task(name="foo"))
+    client.create_task(name="foo")
     mock_rest.create_task.assert_called_once_with(Task(name="foo"))
 
 
 def test_create_task_does_not_start_task(
     client: BlueapiClient,
     mock_rest: Mock,
 ):
-    client.create_task(task=Task(name="foo"))
+    client.create_task(name="foo")
     mock_rest.update_worker_task.assert_not_called()
 
 
@@ -218,7 +218,7 @@ 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(Task(name="baz"))
+    client.create_and_start_task(name="baz")
     mock_rest.create_task.assert_called_once_with(Task(name="baz"))
     mock_rest.update_worker_task.assert_called_once_with(WorkerTask(task_id="baz"))
 
@@ -229,7 +229,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(Task(name="baz"))
+        client.create_and_start_task(name="baz")
 
 
 def test_create_and_start_task_fails_if_task_id_is_wrong(
@@ -239,7 +239,7 @@ def test_create_and_start_task_fails_if_task_id_is_wrong(
     mock_rest.create_task.return_value = TaskResponse(task_id="baz")
     mock_rest.update_worker_task.return_value = TaskResponse(task_id="bar")
     with pytest.raises(BlueskyRemoteControlError):
-        client.create_and_start_task(Task(name="baz"))
+        client.create_and_start_task(name="baz")
 
 
 def test_create_and_start_task_fails_if_task_start_fails(
@@ -249,7 +249,7 @@ def test_create_and_start_task_fails_if_task_start_fails(
     mock_rest.create_task.return_value = TaskResponse(task_id="baz")
     mock_rest.update_worker_task.side_effect = BlueskyRemoteControlError("No can do")
     with pytest.raises(BlueskyRemoteControlError):
-        client.create_and_start_task(Task(name="baz"))
+        client.create_and_start_task(name="baz")
 
 
 def test_get_environment(client: BlueapiClient):
@@ -384,7 +384,7 @@ def test_cannot_run_task_without_message_bus(client: BlueapiClient):
         MissingStompConfiguration,
         match="Stomp configuration required to run plans is missing or disabled",
     ):
-        client.run_task(Task(name="foo"))
+        client.run_task(name="foo")
 
 
 def test_run_task_sets_up_control(
@@ -398,7 +398,7 @@ 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(Task(name="foo"))
+    client_with_events.run_task(name="foo")
     mock_rest.create_task.assert_called_once_with(Task(name="foo"))
     mock_rest.update_worker_task.assert_called_once_with(WorkerTask(task_id="foo"))
 
@@ -417,7 +417,7 @@ def test_run_task_fails_on_failing_event(
 
     on_event = Mock()
     with pytest.raises(BlueskyStreamingError):
-        client_with_events.run_task(Task(name="foo"), on_event=on_event)
+        client_with_events.run_task(name="foo", on_event=on_event)
 
     on_event.assert_called_with(FAILED_EVENT)
 
@@ -456,7 +456,7 @@ def callback(on_event: Callable[[AnyEvent, MessageContext], None]):
     mock_events.subscribe_to_all_events = callback  # type: ignore
 
     mock_on_event = Mock()
-    client_with_events.run_task(Task(name="foo"), on_event=mock_on_event)
+    client_with_events.run_task(name="foo", on_event=mock_on_event)
 
     assert mock_on_event.mock_calls == [call(test_event), call(COMPLETE_EVENT)]
 
@@ -495,7 +495,7 @@ def callback(on_event: Callable[[AnyEvent, MessageContext], None]):
     mock_events.subscribe_to_all_events = callback
 
     mock_on_event = Mock()
-    client_with_events.run_task(Task(name="foo"), on_event=mock_on_event)
+    client_with_events.run_task(name="foo", on_event=mock_on_event)
 
     mock_on_event.assert_called_once_with(COMPLETE_EVENT)
 
@@ -543,8 +543,8 @@ def test_create_task_span_ok(
     client: BlueapiClient,
     mock_rest: Mock,
 ):
-    with asserting_span_exporter(exporter, "create_task", "task"):
-        client.create_task(task=Task(name="foo"))
+    with asserting_span_exporter(exporter, "create_task", "name", "parameters"):
+        client.create_task(name="foo")
 
 
 def test_clear_task_span_ok(
@@ -579,8 +579,10 @@ def test_create_and_start_task_span_ok(
 ):
     mock_rest.create_task.return_value = TaskResponse(task_id="baz")
     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(Task(name="baz"))
+    with asserting_span_exporter(
+        exporter, "create_and_start_task", "name", "parameters"
+    ):
+        client.create_and_start_task(name="baz")
 
 
 def test_get_environment_span_ok(
@@ -644,4 +646,4 @@ 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(Task(name="foo"))
+            client.run_task(name="foo")

tests/unit_tests/test_cli.py

@@ -569,13 +569,13 @@ def test_error_handling(exception, error_message, runner: CliRunner):
 
 
 @pytest.mark.parametrize(
-    "params, error",
+    "params, error, code",
     [
-        ("{", "Parameters are not valid JSON"),
-        ("[]", ""),
+        ("{", "Invalid value for '[PARAMETERS]'", 2),
+        ("[]", "Incorrect parameters supplied", 1),
     ],
 )
-def test_run_task_parsing_errors(params: str, error: str, runner: CliRunner):
+def test_run_task_parsing_errors(params: str, error: str, code: int, runner: CliRunner):
     result = runner.invoke(
         main,
         [
@@ -587,8 +587,8 @@ def test_run_task_parsing_errors(params: str, error: str, runner: CliRunner):
             params,
         ],
     )
-    assert result.stderr.startswith("Error: " + error)
-    assert result.exit_code == 1
+    assert ("Error: " + error) in result.stderr
+    assert result.exit_code == code
 
 
 def test_device_output_formatting():