diff --git a/spyne/const/__init__.py b/spyne/const/__init__.py index eb5ca9043..8f4f6122b 100644 --- a/spyne/const/__init__.py +++ b/spyne/const/__init__.py @@ -76,6 +76,13 @@ """Warn about duplicate faultcodes in all Fault subclasses globally. Only works when CODE class attribute is set for every Fault subclass.""" +READ_ANNOTATIONS = True +"""Whether to use the type annotations as a mechanism of constructing the rpc call +""" +READ_ANNOTATIONS_WARNING = ('Using types in @rpc will be depreciated in future releases, ' + 'please use python type annotations for both your request and response', + DeprecationWarning, + stacklevel=2) def add_request_suffix(string): """Concatenates REQUEST_SUFFIX to end of string""" diff --git a/spyne/decorator.py b/spyne/decorator.py index 38ad8ccf7..dc1ad6b00 100644 --- a/spyne/decorator.py +++ b/spyne/decorator.py @@ -26,12 +26,16 @@ decorator is a simple example of this. """ +from functools import wraps +import inspect +import warnings import spyne.const.xml from copy import copy from inspect import isclass from spyne import MethodDescriptor +import spyne.const # Empty means empty input, bare output. Doesn't say anything about response # being empty @@ -298,7 +302,7 @@ def _get_event_managers(kparams): return _event_managers if _event_managers is not None else [] -def rpc(*params, **kparams): +def _rpc(*params, **kparams): """Method decorator to tag a method as a remote procedure call in a :class:`spyne.service.Service` subclass. @@ -564,9 +568,78 @@ def srpc(*params, **kparams): """ kparams["_no_ctx"] = True - return rpc(*params, **kparams) + return _rpc(*params, **kparams) def mrpc(*params, **kparams): kparams["_no_self"] = False - return rpc(*params, **kparams) + return _rpc(*params, **kparams) + + +def rpc(*args, **kwargs): + no_args = False + if len(args) == 1 and not kwargs and callable(args[0]): + # Called without args + func = args[0] + no_args = True + + if not spyne.const.READ_ANNOTATIONS: + warnings.warn(*spyne.const.READ_ANNOTATIONS_WARNING) + return _rpc + + def _annotated_rpc(func): + inputs = [] + definition = inspect.signature(func) + missing_type_annotations = [] + + input_type_annotations = [ + _param.annotation for _param in definition.parameters.values() + if _param.annotation is not inspect._empty + ] + is_annotated = definition.return_annotation is not inspect._empty \ + or input_type_annotations + + if is_annotated and args and inspect.isclass(args[0]): + raise ValueError( + "*params must be empty when type annotations are used" + ) + + if is_annotated and "_returns" in kwargs: + raise ValueError( + "_returns must be omitted when type annotations are used. " + "Please annotate the return type" + ) + + if not is_annotated: + warnings.warn(*spyne.const.READ_ANNOTATIONS_WARNING) + return _rpc(*args, *kwargs) + + for param_name, param_type in definition.parameters.items(): + if param_name in ("self", "ctx"): + continue + if param_type.annotation is inspect._empty: + missing_type_annotations.append(param_name) + inputs.append(param_type.annotation) + + if missing_type_annotations: + caller = inspect.getframeinfo(inspect.stack()[2][0]) + raise ValueError( + f"{caller.filename}:{caller.lineno} - " + "Missing type annotation for the parameters: " + f"{missing_type_annotations}" + ) + + if definition.return_annotation is not inspect._empty: + new_func = _rpc( + *inputs, _returns=definition.return_annotation, **kwargs + )(func) + else: + new_func = _rpc(*inputs, **kwargs)(func) + + @wraps(new_func) + def wrapper(*args, **kwargs): + return new_func(*args, **kwargs) + + return wrapper + + return _annotated_rpc(func) if no_args else _annotated_rpc \ No newline at end of file diff --git a/spyne/test/test_service.py b/spyne/test/test_service.py index ac09aa3d1..cdf1626cd 100755 --- a/spyne/test/test_service.py +++ b/spyne/test/test_service.py @@ -505,6 +505,72 @@ def method(ctx): raise Exception("Must fail with: " "'SelfReference can't be used inside @rpc and its ilk'") + def test_annotated_rpc_works_with_no_kwargs(self): + class someCallResponse(ComplexModel): + __namespace__ = 'tns' + s = String + + class SomeService(Service): + @rpc + def someCall(ctx, x: someCallResponse) -> Array(String): + return ['abc', 'def'] + + def test_annotated_rpc_works_with_kwargs(self): + class someCallResponse(ComplexModel): + __namespace__ = 'tns' + s = String + + class SomeService(Service): + @rpc(_is_async=True) + def someCall(ctx, x: someCallResponse) -> Array(String): + return ['abc', 'def'] + + + def test_annotated_rpc_works_with_no_response_works(self): + class someCallResponse(ComplexModel): + __namespace__ = 'tns' + s = String + + class SomeService(Service): + @rpc(_is_async=True) + def someCall(ctx, x: someCallResponse): + return ['abc', 'def'] + + def test_annotated_rpc_works_with__returns_kwarg_raises(self): + class someCallResponse(ComplexModel): + __namespace__ = 'tns' + s = String + + expected_message = "_returns must be omitted when type annotations " \ + "are used. Please annotate the return type" + try: + class SomeService(Service): + @rpc(_returns=Array(String)) + def someCall(ctx, x: someCallResponse) -> Array(String): + return ['abc', 'def'] + except ValueError as e: + assert str(e) == expected_message + else: + raise Exception(f"Must fail with: ValueError('{expected_message}'") + + def test_annotated_rpc_works_with_missing_type_annotation_raises(self): + class someCallResponse(ComplexModel): + __namespace__ = 'tns' + s = String + + + e_sub_message = "Missing type annotation for the parameters: ['y']" + try: + class SomeService(Service): + @rpc(_is_async=True) + def someCall(ctx, x: someCallResponse, y) -> Array(String): + return ['abc', 'def'] + except ValueError as e: + assert e_sub_message in str(e) + else: + raise Exception(f"Must fail with: ValueError('{e_sub_message}'") + + if __name__ == '__main__': unittest.main()