Support operation functions with **kwargs

connexion/decorators/parameter.py

@@ -25,12 +25,15 @@ def get_function_arguments(function):  # pragma: no cover
     Returns the list of arguments of a function
 
     :type function: Callable
-    :rtype: list[str]
+    :rtype: tuple[list[str], bool]
     """
     if six.PY3:
-        return list(inspect.signature(function).parameters)
+        parameters = inspect.signature(function).parameters
+        return (list([name for name, p in parameters.items() if p.kind not in (p.VAR_POSITIONAL, p.VAR_KEYWORD)]),
+                any(p.kind == p.VAR_KEYWORD for p in parameters.values()))
     else:
-        return inspect.getargspec(function).args
+        argspec = inspect.getargspec(function)
+        return argspec.args, bool(argspec.keywords)
 
 
 def make_type(value, type):
@@ -71,7 +74,7 @@ def parameter_to_arg(parameters, function):
                   for parameter in parameters if parameter['in'] == 'formData'}
     path_types = {parameter['name']: parameter
                   for parameter in parameters if parameter['in'] == 'path'}
-    arguments = get_function_arguments(function)
+    arguments, has_kwargs = get_function_arguments(function)
     default_query_params = {param['name']: param['default']
                             for param in parameters if param['in'] == 'query' and 'default' in param}
     default_form_params = {param['name']: param['default']
@@ -96,17 +99,17 @@ def wrapper(*args, **kwargs):
                                                  path_param_definitions)
 
         # Add body parameters
-        if body_name not in arguments:
+        if not has_kwargs and body_name not in arguments:
             logger.debug("Body parameter '%s' not in function arguments", body_name)
-        else:
+        elif body_name:
             logger.debug("Body parameter '%s' in function arguments", body_name)
             kwargs[body_name] = request_body
 
         # Add query parameters
         query_arguments = copy.deepcopy(default_query_params)
         query_arguments.update(flask.request.args.items())
         for key, value in query_arguments.items():
-            if key not in arguments:
+            if not has_kwargs and key not in arguments:
                 logger.debug("Query Parameter '%s' not in function arguments", key)
             else:
                 logger.debug("Query Parameter '%s' in function arguments", key)
@@ -122,7 +125,7 @@ def wrapper(*args, **kwargs):
         form_arguments = copy.deepcopy(default_form_params)
         form_arguments.update(flask.request.form.items())
         for key, value in form_arguments.items():
-            if key not in arguments:
+            if not has_kwargs and key not in arguments:
                 logger.debug("FormData parameter '%s' not in function arguments", key)
             else:
                 logger.debug("FormData parameter '%s' in function arguments", key)
@@ -136,7 +139,7 @@ def wrapper(*args, **kwargs):
         # Add file parameters
         file_arguments = flask.request.files
         for key, value in file_arguments.items():
-            if key not in arguments:
+            if not has_kwargs and key not in arguments:
                 logger.debug("File parameter (formData) '%s' not in function arguments", key)
             else:
                 logger.debug("File parameter (formData) '%s' in function arguments", key)

tests/api/test_parameters.py

@@ -276,3 +276,16 @@ def test_nullable_parameter(simple_app):
 
     resp = app_client.put('/v1.0/nullable-parameters', data="None")
     assert json.loads(resp.data.decode()) == 'it was None'
+
+
+def test_args_kwargs(simple_app):
+    app_client = simple_app.app.test_client()
+    resp = app_client.get('/v1.0/query-params-as-kwargs')
+    assert resp.status_code == 200
+    assert json.loads(resp.data.decode()) == {}
+
+    resp = app_client.get('/v1.0/query-params-as-kwargs?foo=a&bar=b')
+    assert resp.status_code == 200
+    assert json.loads(resp.data.decode()) == {'foo': 'a'}
+
+

tests/fakeapi/hello.py

@@ -1,5 +1,7 @@
 #!/usr/bin/env python3
 
+import json
+
 from connexion import NoContent, problem, request
 from flask import redirect
 
@@ -345,3 +347,7 @@ def unordered_params_response(first, path_param, second):
 
 def more_than_one_scope_defined():
     return "OK"
+
+
+def test_args_kwargs(*args, **kwargs):
+    return kwargs

tests/fixtures/simple/swagger.yaml

@@ -614,6 +614,22 @@ paths:
           schema:
             type: string
 
+  /query-params-as-kwargs:
+    get:
+      operationId: fakeapi.hello.test_args_kwargs
+      produces:
+        - application/json
+      parameters:
+        - name: foo
+          description: Just a testing parameter.
+          in: query
+          type: string
+      responses:
+        200:
+          description: Return kwargs
+          schema:
+            type: object
+
 
 definitions:
   new_stack:

tests/test_parameter.py

@@ -13,12 +13,20 @@ def stub_function(foo, bar):
     pass
 
 
+def kwargs_function(*args, **kwargs):
+    pass
+
+
 def test_get_proper_argument_list():
     """Test get the proper argument list of the decorated function."""
 
-    assert len(get_function_arguments(stub_function)) == 2
-    assert get_function_arguments(stub_function) == ['foo', 'bar']
+    assert len(get_function_arguments(stub_function)[0]) == 2
+    assert get_function_arguments(stub_function) == (['foo', 'bar'], False)
 
     decorated_stub_function = the_decorator(stub_function)
-    assert len(get_function_arguments(decorated_stub_function)) == 2
-    assert get_function_arguments(decorated_stub_function) == ['foo', 'bar']
+    assert len(get_function_arguments(decorated_stub_function)[0]) == 2
+    assert get_function_arguments(decorated_stub_function) == (['foo', 'bar'], False)
+
+
+def test_get_kwargs():
+    assert get_function_arguments(kwargs_function) == ([], True)