First draft to show how to make syntactically nicer with type annotations

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"""

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

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()