"""Decorators to provide argument validation to a Tango command."""
import functools
import inspect
import json
import logging
from collections.abc import Callable
from typing import Any, overload
import jsonschema
from .type_hints import JSONData
module_logger = logging.getLogger(__name__)
_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.
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.
"""
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:
logger: logging.Logger = getattr(self, "logger", module_logger)
# 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_method.__name__}_SCHEMA", default_schema
)
elif isinstance(schema, str):
try:
resolved_schema = getattr(self, schema)
except AttributeError as e:
logger.error("@validate_json_args: Schema not found: %s", e)
raise e
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:
logger.exception(
"@validate_json_args: The given argument for '%s' is not valid JSON: "
"'%s'",
command_method.__name__,
json_str,
)
raise
try:
jsonschema.validate(decoded_dict, resolved_schema)
except jsonschema.ValidationError:
logger.exception(
"@validate_json_args: The given argument for '%s' failed JSON schema "
"validation: '%s'",
command_method.__name__,
json_str,
)
raise
if not isinstance(decoded_dict, dict):
message = (
"@validate_json_args must be used with a JSON dictionary of arguments."
)
logger.error(message)
raise TypeError(message)
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