Swagger schema partial patching. Swagger schema validation.

.gitignore

@@ -132,3 +132,6 @@ ENV/
 
 # Pycharm
 .idea/
+
+# pytest
+.pytest_cache/

.travis.yml

@@ -1,7 +1,6 @@
 language: python
 sudo: false
 python:
-  - "3.4"
   - "3.5"
   - "3.6"
 
@@ -30,4 +29,4 @@ deploy:
 #    script: deploy/pypi.sh
 #    skip_cleanup: true
 #    on:
-#      branch: master
+#      branch: master

aiohttp_swagger/__init__.py

@@ -1,17 +1,33 @@
 import asyncio
-from os.path import abspath, dirname, join
+from os.path import (
+    abspath,
+    dirname,
+    join,
+)
 from types import FunctionType
 
 from aiohttp import web
 
-from .helpers import (generate_doc_from_each_end_point,
-                      load_doc_from_yaml_file, swagger_path)
+from .helpers import (
+    generate_doc_from_each_end_point,
+    load_doc_from_yaml_file,
+    load_doc_from_yaml_str,
+    swagger_path,
+    swagger_validation,
+    add_swagger_validation,
+)
 
 try:
     import ujson as json
 except ImportError:
     import json
 
+__all__ = (
+    "setup_swagger",
+    "swagger_path",
+    "swagger_validation",
+)
+
 
 @asyncio.coroutine
 def _swagger_home(request):
@@ -38,32 +54,57 @@ def _swagger_def(request):
 def setup_swagger(app: web.Application,
                   *,
                   swagger_from_file: str = None,
+                  swagger_from_str: str = None,
                   swagger_url: str = "/api/doc",
                   api_base_url: str = "/",
+                  swagger_validator_url: str = "",
                   description: str = "Swagger API definition",
                   api_version: str = "1.0.0",
                   title: str = "Swagger API",
                   contact: str = "",
                   swagger_home_decor: FunctionType = None,
                   swagger_def_decor: FunctionType = None,
+                  swagger_merge_with_file: bool = False,
+                  swagger_validate_schema: bool = False,
                   swagger_info: dict = None):
     _swagger_url = ("/{}".format(swagger_url)
                     if not swagger_url.startswith("/")
                     else swagger_url)
     _base_swagger_url = _swagger_url.rstrip('/')
     _swagger_def_url = '{}/swagger.json'.format(_base_swagger_url)
 
-    # Build Swagget Info
+    # Build Swagger Info
     if swagger_info is None:
-        if swagger_from_file:
-            swagger_info = load_doc_from_yaml_file(swagger_from_file)
+        if swagger_from_file or swagger_from_str:
+            if swagger_from_file:
+                swagger_info = load_doc_from_yaml_file(swagger_from_file)
+            elif swagger_from_str:
+                swagger_info = load_doc_from_yaml_str(swagger_from_str)
+            if swagger_merge_with_file:
+                swagger_end_points_info = generate_doc_from_each_end_point(
+                    app, api_base_url=api_base_url, description=description,
+                    api_version=api_version, title=title, contact=contact
+                )
+                paths = swagger_end_points_info.pop('paths', None)
+                swagger_info.update(swagger_end_points_info)
+                if paths is not None:
+                    if 'paths' not in swagger_info:
+                        swagger_info['paths'] = {}
+                    for ph, description in paths.items():
+                        for method, desc in description.items():
+                            if ph not in swagger_info['paths']:
+                                swagger_info['paths'][ph] = {}
+                            swagger_info['paths'][ph][method] = desc
         else:
             swagger_info = generate_doc_from_each_end_point(
                 app, api_base_url=api_base_url, description=description,
                 api_version=api_version, title=title, contact=contact
             )
-    else:
-        swagger_info = json.dumps(swagger_info)
+
+    if swagger_validate_schema:
+        add_swagger_validation(app, swagger_info)
+
+    swagger_info = json.dumps(swagger_info)
 
     _swagger_home_func = _swagger_home
     _swagger_def_func = _swagger_def
@@ -91,11 +132,9 @@ def setup_swagger(app: web.Application,
     with open(join(STATIC_PATH, "index.html"), "r") as f:
         app["SWAGGER_TEMPLATE_CONTENT"] = (
             f.read()
-            .replace("##SWAGGER_CONFIG##", '/{}{}'.
+            .replace("##SWAGGER_CONFIG##", '{}{}'.
                      format(api_base_url.lstrip('/'), _swagger_def_url))
-            .replace("##STATIC_PATH##", '/{}{}'.
+            .replace("##STATIC_PATH##", '{}{}'.
                      format(api_base_url.lstrip('/'), statics_path))
+            .replace("##SWAGGER_VALIDATOR_URL##", swagger_validator_url)
         )
-
-
-__all__ = ("setup_swagger", "swagger_path")

aiohttp_swagger/helpers/__init__.py

@@ -1,2 +1,3 @@
 from .builders import *  # noqa
 from .decorators import *  # noqa
+from .validation import *  # noqa

aiohttp_swagger/helpers/builders.py

@@ -1,22 +1,35 @@
+import logging
+from typing import (
+    MutableMapping,
+    Mapping,
+    TextIO,
+)
 from collections import defaultdict
-from os.path import abspath, dirname, join
+from os.path import (
+    abspath,
+    dirname,
+    join,
+)
 
 import yaml
 from aiohttp import web
 from aiohttp.hdrs import METH_ANY, METH_ALL
 from jinja2 import Template
-
 try:
     import ujson as json
-except ImportError: # pragma: no cover
+except ImportError:  # pragma: no cover
     import json
 
+from .validation import validate_decorator
+
 
 SWAGGER_TEMPLATE = abspath(join(dirname(__file__), "..", "templates"))
 
 
-def _extract_swagger_docs(end_point_doc, method="get"):
-    # Find Swagger start point in doc
+def _extract_swagger_docs(end_point_doc: str) -> Mapping:
+    """
+    Find Swagger start point in doc.
+    """
     end_point_swagger_start = 0
     for i, doc_line in enumerate(end_point_doc):
         if "---" in doc_line:
@@ -34,39 +47,41 @@ def _extract_swagger_docs(end_point_doc, method="get"):
                            "from docstring ⚠",
             "tags": ["Invalid Swagger"]
         }
-    return {method: end_point_swagger_doc}
+    return end_point_swagger_doc
+
 
 def _build_doc_from_func_doc(route):
 
     out = {}
 
     if issubclass(route.handler, web.View) and route.method == METH_ANY:
         method_names = {
-            attr for attr in dir(route.handler) \
+            attr for attr in dir(route.handler)
             if attr.upper() in METH_ALL
         }
         for method_name in method_names:
             method = getattr(route.handler, method_name)
             if method.__doc__ is not None and "---" in method.__doc__:
                 end_point_doc = method.__doc__.splitlines()
-                out.update(_extract_swagger_docs(end_point_doc, method=method_name))
+                out[method_name] = _extract_swagger_docs(end_point_doc)
 
     else:
         try:
             end_point_doc = route.handler.__doc__.splitlines()
         except AttributeError:
             return {}
-        out.update(_extract_swagger_docs(end_point_doc))
+        out[route.method.lower()] = _extract_swagger_docs(end_point_doc)
     return out
 
+
 def generate_doc_from_each_end_point(
         app: web.Application,
         *,
         api_base_url: str = "/",
         description: str = "Swagger API definition",
         api_version: str = "1.0.0",
         title: str = "Swagger API",
-        contact: str = ""):
+        contact: str = "") -> MutableMapping:
     # Clean description
     _start_desc = 0
     for i, word in enumerate(description):
@@ -92,8 +107,6 @@ def generate_doc_from_each_end_point(
 
     for route in app.router.routes():
 
-        end_point_doc = None
-
         # If route has a external link to doc, we use it, not function doc
         if getattr(route.handler, "swagger_file", False):
             try:
@@ -133,13 +146,66 @@ def generate_doc_from_each_end_point(
                 url = url_info.get("formatter")
 
             swagger["paths"][url].update(end_point_doc)
+    return swagger
+
 
-    return json.dumps(swagger)
+def load_doc_from_yaml_file(doc_path: str) -> MutableMapping:
+    return yaml.load(open(doc_path, "r").read())
 
 
-def load_doc_from_yaml_file(doc_path: str):
-    loaded_yaml = yaml.load(open(doc_path, "r").read())
-    return json.dumps(loaded_yaml)
+def load_doc_from_yaml_file_obj(doc: TextIO) -> MutableMapping:
+    return yaml.load(doc.read())
 
 
-__all__ = ("generate_doc_from_each_end_point", "load_doc_from_yaml_file")
+def load_doc_from_yaml_str(doc: str) -> MutableMapping:
+    return yaml.load(doc)
+
+
+def add_swagger_validation(app, swagger_info: Mapping):
+    for route in app.router.routes():
+        method = route.method.lower()
+        handler = route.handler
+        url_info = route.get_info()
+        url = url_info.get('path') or url_info.get('formatter')
+
+        if method != '*':
+            swagger_endpoint_info_for_method = \
+                swagger_info['paths'].get(url, {}).get(method)
+            swagger_endpoint_info = \
+                {method: swagger_endpoint_info_for_method} if \
+                swagger_endpoint_info_for_method is not None else {}
+        else:
+            # all methods
+            swagger_endpoint_info = swagger_info['paths'].get(url, {})
+        for method, info in swagger_endpoint_info.items():
+            logging.debug(
+                'Added validation for method: {}. Path: {}'.
+                format(method.upper(), url)
+            )
+            if issubclass(handler, web.View) and route.method == METH_ANY:
+                # whole class validation
+                should_be_validated = getattr(handler, 'validation', False)
+                cls_method = getattr(handler, method, None)
+                if cls_method is not None:
+                    if not should_be_validated:
+                        # method validation
+                        should_be_validated = \
+                            getattr(handler, 'validation', False)
+                    if should_be_validated:
+                        new_cls_method = \
+                            validate_decorator(swagger_info, info)(cls_method)
+                        setattr(handler, method, new_cls_method)
+            else:
+                should_be_validated = getattr(handler, 'validation', False)
+                if should_be_validated:
+                    route._handler = \
+                        validate_decorator(swagger_info, info)(handler)
+
+
+__all__ = (
+    "generate_doc_from_each_end_point",
+    "load_doc_from_yaml_file",
+    "load_doc_from_yaml_str",
+    "load_doc_from_yaml_file_obj",
+    "add_swagger_validation",
+)

aiohttp_swagger/helpers/decorators.py

@@ -1,7 +1,27 @@
-class swagger_path(object):
+from functools import partial
+from inspect import isfunction, isclass
+
+__all__ = (
+    'swagger_path',
+    'swagger_validation',
+)
+
+
+class swagger_path:
+
     def __init__(self, swagger_file):
         self.swagger_file = swagger_file
 
     def __call__(self, f):
         f.swagger_file = self.swagger_file
         return f
+
+
+def swagger_validation(func=None, *, validation=True):
+
+    if func is None or not (isfunction(func) or isclass(func)):
+        validation = func
+        return partial(swagger_validation, validation=validation)
+
+    func.validation = validation
+    return func