Multiple fixes and improvements

sanic_openapi/__init__.py

@@ -1,13 +1,14 @@
 from .openapi2 import doc, openapi2_blueprint
-from .openapi3 import openapi, openapi3_blueprint
+from .openapi3 import openapi, openapi3_blueprint, specification
 
 swagger_blueprint = openapi2_blueprint
 
-__version__ = "21.3.3"
+__version__ = "21.6.0"
 __all__ = [
     "openapi2_blueprint",
     "swagger_blueprint",
     "openapi3_blueprint",
     "openapi",
+    "specification",
     "doc",
 ]

sanic_openapi/openapi2/blueprint.py

@@ -54,6 +54,8 @@ def build_spec(app, loop):
         for blueprint_name, handler in get_blueprinted_routes(app):
             route_spec = route_specs[handler]
             route_spec.blueprint = blueprint_name
+            if route_spec.exclude:
+                continue
             if not route_spec.tags:
                 route_spec.tags.append(blueprint_name)
 
@@ -195,7 +197,9 @@ def build_spec(app, loop):
                 methods[_method.lower()] = endpoint
 
             if methods:
-                paths[uri] = methods
+                if uri not in paths:
+                    paths[uri] = {}
+                paths[uri].update(methods)
 
         # --------------------------------------------------------------- #
         # Definitions

sanic_openapi/openapi3/__init__.py

@@ -3,15 +3,23 @@
 """
 
 from collections import defaultdict
-
+from typing import Dict, TypeVar
 from .builders import OperationBuilder, SpecificationBuilder
 
+try:
+    from sanic.models.handler_types import RouteHandler
+except ImportError:
+    RouteHandler = TypeVar("RouteHandler")  # type: ignore
+
 # Static datastores, which get added to via the oas3.openapi decorators,
 # and then read from in the blueprint generation
 
-operations = defaultdict(OperationBuilder)
+operations: Dict[RouteHandler, OperationBuilder] = defaultdict(
+    OperationBuilder
+)
 specification = SpecificationBuilder()
 
+
 from .blueprint import blueprint_factory  # noqa
 
 

sanic_openapi/openapi3/blueprint.py

@@ -76,6 +76,10 @@ def build_spec(app, loop):
                 if hasattr(_handler, "view_class"):
                     _handler = getattr(_handler.view_class, method.lower())
                 operation = operations[_handler]
+
+                if operation._exclude:
+                    continue
+
                 docstring = inspect.getdoc(_handler)
 
                 if docstring:
@@ -107,19 +111,19 @@ def add_static_info_to_spec_from_config(app, specification):
 
     Modifies specification in-place and returns None
     """
-    specification.describe(
+    specification._do_describe(
         getattr(app.config, "API_TITLE", "API"),
         getattr(app.config, "API_VERSION", "1.0.0"),
         getattr(app.config, "API_DESCRIPTION", None),
         getattr(app.config, "API_TERMS_OF_SERVICE", None),
     )
 
-    specification.license(
+    specification._do_license(
         getattr(app.config, "API_LICENSE_NAME", None),
         getattr(app.config, "API_LICENSE_URL", None),
     )
 
-    specification.contact(
+    specification._do_contact(
         getattr(app.config, "API_CONTACT_NAME", None),
         getattr(app.config, "API_CONTACT_URL", None),
         getattr(app.config, "API_CONTACT_EMAIL", None),

sanic_openapi/openapi3/builders.py

@@ -11,6 +11,7 @@
 from ..utils import remove_nulls, remove_nulls_from_kwargs
 from .definitions import (
     Any,
+    Components,
     Contact,
     Dict,
     ExternalDocumentation,
@@ -47,6 +48,7 @@ def __init__(self):
         self.parameters = []
         self.responses = {}
         self._autodoc = None
+        self._exclude = False
 
     def name(self, value: str):
         self.operationId = value
@@ -108,6 +110,9 @@ def autodoc(self, docstring: str):
         y = YamlStyleParametersParser(docstring)
         self._autodoc = y.to_openAPI_3()
 
+    def exclude(self, flag: bool = True):
+        self._exclude = flag
+
 
 class SpecificationBuilder:
     _urls: List[str]
@@ -119,14 +124,24 @@ class SpecificationBuilder:
     _license: License
     _paths: Dict[str, Dict[str, OperationBuilder]]
     _tags: Dict[str, Tag]
+    _components: Dict[str, Any]
+    _servers: List[Server]
     # _components: ComponentsBuilder
     # deliberately not included
 
     def __init__(self):
+        self._components = defaultdict(dict)
+        self._contact = None
+        self._description = None
+        self._external = None
+        self._license = None
         self._paths = defaultdict(dict)
+        self._servers = []
         self._tags = {}
-        self._license = None
+        self._terms = None
+        self._title = None
         self._urls = []
+        self._version = None
 
     def url(self, value: str):
         self._urls.append(value)
@@ -143,17 +158,45 @@ def describe(
         self._description = description
         self._terms = terms
 
-    def tag(self, name: str, **kwargs):
-        self._tags[name] = Tag(name, **kwargs)
+    def _do_describe(
+        self,
+        title: str,
+        version: str,
+        description: Optional[str] = None,
+        terms: Optional[str] = None,
+    ):
+        if any([self._title, self._version, self._description, self._terms]):
+            return
+        self.describe(title, version, description, terms)
+
+    def tag(self, name: str, description: Optional[str] = None, **kwargs):
+        self._tags[name] = Tag(name, description=description, **kwargs)
+
+    def external(self, url: str, description: Optional[str] = None, **kwargs):
+        self._external = ExternalDocumentation(url, description=description)
 
     def contact(self, name: str = None, url: str = None, email: str = None):
         kwargs = remove_nulls_from_kwargs(name=name, url=url, email=email)
         self._contact = Contact(**kwargs)
 
+    def _do_contact(
+        self, name: str = None, url: str = None, email: str = None
+    ):
+        if self._contact:
+            return
+
+        self.contact(name, url, email)
+
     def license(self, name: str = None, url: str = None):
         if name is not None:
             self._license = License(name, url=url)
 
+    def _do_license(self, name: str = None, url: str = None):
+        if self._license:
+            return
+
+        self.license(name, url)
+
     def operation(self, path: str, method: str, operation: OperationBuilder):
         for _tag in operation.tags:
             if _tag in self._tags.keys():
@@ -163,18 +206,62 @@ def operation(self, path: str, method: str, operation: OperationBuilder):
 
         self._paths[path][method.lower()] = operation
 
+    def add_component(self, location: str, name: str, obj: Any):
+        self._components[location].update({name: obj})
+
+    def raw(self, data):
+        if "info" in data:
+            self.describe(
+                data["info"].get("title"),
+                data["info"].get("version"),
+                data["info"].get("description"),
+                data["info"].get("terms"),
+            )
+
+        if "servers" in data:
+            for server in data["servers"]:
+                self._servers.append(Server(**server))
+
+        if "paths" in data:
+            self._paths.update(data["paths"])
+
+        if "components" in data:
+            for location, component in data["components"].items():
+                self._components[location].update(component)
+
+        if "security" in data:
+            ...
+
+        if "tags" in data:
+            for tag in data["tags"]:
+                self.tag(**tag)
+
+        if "externalDocs" in data:
+            self.external(**data["externalDocs"])
+
     def build(self) -> OpenAPI:
         info = self._build_info()
         paths = self._build_paths()
         tags = self._build_tags()
 
         url_servers = getattr(self, "_urls", None)
-        servers = []
+        servers = self._servers
         if url_servers is not None:
             for url_server in url_servers:
                 servers.append(Server(url=url_server))
 
-        return OpenAPI(info, paths, tags=tags, servers=servers)
+        components = (
+            Components(**self._components) if self._components else None
+        )
+
+        return OpenAPI(
+            info,
+            paths,
+            tags=tags,
+            servers=servers,
+            components=components,
+            externalDocs=self._external,
+        )
 
     def _build_info(self) -> Info:
         kwargs = remove_nulls(
@@ -197,7 +284,10 @@ def _build_paths(self) -> Dict:
 
         for path, operations in self._paths.items():
             paths[path] = PathItem(
-                **{k: v.build() for k, v in operations.items()}
+                **{
+                    k: v if isinstance(v, dict) else v.build()
+                    for k, v in operations.items()
+                }
             )
 
         return paths