"""Decorators to provide argument validation to a Tango command."""
import functools
import inspect
import json
import warnings
from collections.abc import Callable
from typing import Any, overload
import jsonschema
from .faults import ValidateJSONArgsError
from .type_hints import JSONData
_JsonFunction = Callable[[Any, str], Any]
_JsonDecorator = Callable[[Callable[..., Any]], _JsonFunction]
@overload
def validate_json_args(
*,
schema: str | dict[str, JSONData] | None = None,
) -> _JsonDecorator: ...
@overload
def validate_json_args(
command_method: Callable[..., Any],
*,
schema: str | dict[str, JSONData] | None = None,
) -> _JsonFunction: ...
[docs]
def validate_json_args(
command_method: Callable[..., Any] | None = None,
*,
schema: str | dict[str, JSONData] | None = None,
) -> _JsonFunction | _JsonDecorator:
"""
Decorate a Tango command to take a validated JSON string as its single argument.
This decorator takes a function with a collection of keyword arguments and
converts it into a function which takes a single string argument.
The string argument is decoded as JSON and validated against an optional schema.
The resulting dictionary's key-value pairs matches the original function's keyword
arguments.
For long running commands this decorator must always be applied first (innermost)
to allow the LRC to catch validation errors and reject the task.
Only one of the optional keyword parameters for a schema can be used:
:param schema: Provide a JSON schema (dict) to validate the input JSON string
against. Optional keyword.
:param schema_name: Provide a name (str) of a JSON schema object to access in the
parent class (Tango device). Optional keyword.
:raises ValidateJSONArgsError: If either the JSON schema cannot be found, the given
command arg is not valid JSON or a JSON dict, or it failed schema validation.
"""
if command_method is None:
return functools.partial(validate_json_args, schema=schema)
default_schema = {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://example.com/product.schema.json",
"title": "Default schema",
"description": "Validates the item as a dictionary",
"type": "object",
}
@functools.wraps(command_method)
def wrapper(self: Any, json_str: str) -> Any:
command_name = command_method.__name__
# Check if @long_running_command or @submit_lrc_task was already applied
if hasattr(command_method, "__is_long_running__"):
warnings.warn(
f"Incorrect decorator order on '{command_name}' command. "
"'@validate_json_args' should be applied first (innermost)",
UserWarning,
stacklevel=2,
)
# Check the schema arg and resolve a schema to use for validation
if schema is None:
resolved_schema: dict[str, JSONData] = getattr(
self, f"{command_name}_SCHEMA", default_schema
)
elif isinstance(schema, str):
try:
resolved_schema = getattr(self, schema)
except AttributeError as exc:
raise ValidateJSONArgsError(
f"Schema for '{command_name}' not found: {exc}"
)
else:
resolved_schema = schema
# Decode the JSON string, validate it against the schema and return the wrapped
# command method if successful.
try:
decoded_dict = json.loads(json_str)
except json.JSONDecodeError as exc:
raise ValidateJSONArgsError(
f"The given argument for '{command_name}' is not valid JSON: "
f"'{json_str}'\n{exc.msg}"
)
try:
jsonschema.validate(decoded_dict, resolved_schema)
except jsonschema.ValidationError as exc:
raise ValidateJSONArgsError(
exc.message,
validator=exc.validator, # type: ignore[arg-type]
validator_value=exc.validator_value,
instance=exc.instance,
schema=exc.schema,
)
if not isinstance(decoded_dict, dict):
raise ValidateJSONArgsError(
"@validate_json_args must be used with a JSON dictionary of arguments. "
f"Given JSON string: '{decoded_dict}'"
)
return command_method(self, **decoded_dict)
# Change signature to be a single argument of type string so that Pytango creates
# a command with dtype_in=str to match json input.
wrapper_sig = inspect.signature(wrapper)
new_sig = wrapper_sig.replace(
parameters=[
inspect.Parameter("self", kind=inspect.Parameter.POSITIONAL_ONLY),
inspect.Parameter(
"json_str", annotation=str, kind=inspect.Parameter.POSITIONAL_ONLY
),
]
)
wrapper.__signature__ = new_sig # type: ignore[attr-defined]
return wrapper