|
6 | 6 | from functools import wraps |
7 | 7 | from pathlib import Path |
8 | 8 | from pprint import pprint |
9 | | -from typing import Any, cast |
| 9 | +from typing import Any, TypeGuard |
10 | 10 |
|
11 | 11 | import click |
12 | 12 | from bluesky.callbacks.best_effort import BestEffortCallback |
@@ -48,20 +48,25 @@ class ParametersType(ParamType): |
48 | 48 | name = "TaskParameters" |
49 | 49 |
|
50 | 50 | def convert( |
51 | | - self, value: Any, param: Parameter | None, ctx: Context | None |
| 51 | + self, |
| 52 | + value: str | dict[str, Any] | None, |
| 53 | + param: Parameter | None, |
| 54 | + ctx: Context | None, |
52 | 55 | ) -> TaskParameters: |
53 | 56 | if isinstance(value, str): |
54 | 57 | try: |
55 | 58 | params = json.loads(value) |
56 | | - if not isinstance(params, dict) or any( |
57 | | - not isinstance(k, str) for k in params |
58 | | - ): |
59 | | - self.fail("Parameters must be a JSON object") |
60 | | - return cast(TaskParameters, params) |
| 59 | + if is_str_dict(params): |
| 60 | + return params |
| 61 | + self.fail("Parameters must be a JSON object with string keys") |
61 | 62 | except json.JSONDecodeError as jde: |
62 | 63 | self.fail(f"Parameters are not valid JSON: {jde}") |
63 | 64 | else: |
64 | | - return cast(TaskParameters, super().convert(value, param, ctx)) |
| 65 | + return super().convert(value, param, ctx) |
| 66 | + |
| 67 | + |
| 68 | +def is_str_dict(val: Any) -> TypeGuard[TaskParameters]: |
| 69 | + return isinstance(val, dict) and all(isinstance(k, str) for k in val) |
65 | 70 |
|
66 | 71 |
|
67 | 72 | @click.group(invoke_without_command=True) |
|
0 commit comments