Added swagger_merge_with_file parameter.

Allows redefine the swagger description in the code for needed endpoints

Author: trezorg

.gitignore

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

aiohttp_swagger/__init__.py

@@ -1,11 +1,18 @@
 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,
+    swagger_path,
+)
 
 try:
     import ujson as json
@@ -47,24 +54,44 @@ def setup_swagger(app: web.Application,
                   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_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:
+        pass
+
+    swagger_info = json.dumps(swagger_info)
 
     _swagger_home_func = _swagger_home
     _swagger_def_func = _swagger_def

aiohttp_swagger/helpers/builders.py

@@ -1,5 +1,13 @@
+from typing import (
+    MutableMapping,
+    Mapping,
+)
 from collections import defaultdict
-from os.path import abspath, dirname, join
+from os.path import (
+    abspath,
+    dirname,
+    join,
+)
 
 import yaml
 from aiohttp import web
@@ -8,7 +16,7 @@
 
 try:
     import ujson as json
-except ImportError: # pragma: no cover
+except ImportError:  # pragma: no cover
     import json
 
 
@@ -36,37 +44,41 @@ def _extract_swagger_docs(end_point_doc, method="get"):
         }
     return {method: 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.update(
+                    _extract_swagger_docs(end_point_doc, method=method_name))
 
     else:
         try:
             end_point_doc = route.handler.__doc__.splitlines()
         except AttributeError:
             return {}
-        out.update(_extract_swagger_docs(end_point_doc))
+        out.update(_extract_swagger_docs(
+            end_point_doc, method=route.method.lower()))
     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 +104,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 +143,14 @@ def generate_doc_from_each_end_point(
                 url = url_info.get("formatter")
 
             swagger["paths"][url].update(end_point_doc)
-
-    return json.dumps(swagger)
+    return swagger
 
 
-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(doc_path: str) -> MutableMapping:
+    return yaml.load(open(doc_path, "r").read())
 
 
-__all__ = ("generate_doc_from_each_end_point", "load_doc_from_yaml_file")
+__all__ = (
+    "generate_doc_from_each_end_point",
+    "load_doc_from_yaml_file"
+)

doc/source/customizing.rst

@@ -168,6 +168,50 @@ Global Swagger YAML
 
     web.run_app(app, host="127.0.0.1")
 
+
+:samp:`aiohttp-swagger` also allow to build an external YAML Swagger file
+      and merge swagger endpoint definitions to it:
+
+.. code-block:: python
+
+    from aiohttp import web
+    from aiohttp_swagger import *
+
+    async def ping(request):
+        """
+        ---
+        tags:
+        - user
+        summary: Create user
+        description: This can only be done by the logged in user.
+        operationId: examples.api.api.createUser
+        produces:
+        - application/json
+        parameters:
+        - in: body
+          name: body
+          description: Created user object
+          required: false
+
+        responses:
+          "201":
+            description: successful operation
+        """
+        return web.Response(text="pong")
+
+    app = web.Application()
+
+    app.router.add_route('GET', "/ping", ping)
+
+    setup_swagger(
+        app,
+        swagger_from_file="example_swagger.yaml",   # <-- Loaded Swagger from external YAML file
+        swagger_merge_with_file=True  # <-- Merge
+    )
+
+    web.run_app(app, host="127.0.0.1")
+
+
 Nested applications
 +++++++++++++++++++
 

tests/conftest.py

@@ -0,0 +1,22 @@
+from os.path import (
+    abspath,
+    dirname,
+    join,
+)
+
+import yaml
+import pytest
+
+
+@pytest.fixture
+def swagger_file():
+    tests_path = abspath(join(dirname(__file__)))
+    return join(tests_path, "data", "example_swagger.yaml")
+
+
+@pytest.fixture
+def swagger_info():
+    filename = abspath(join(dirname(__file__))) + "/data/example_swagger.yaml"
+    return yaml.load(open(filename).read())
+
+