Merge pull request #210 from artcg/ruck3-bugfix

OAS3 support for sanic-openapi3

sanic_openapi/__init__.py

@@ -1,4 +1,4 @@
-from .swagger import swagger_blueprint
+from .swagger import oas3_blueprint, swagger_blueprint
 
 __version__ = "0.6.2"
-__all__ = ["swagger_blueprint"]
+__all__ = ["swagger_blueprint", "oas3_blueprint"]

sanic_openapi/doc.py

@@ -136,11 +136,7 @@ def definition(self):
             "properties": {
                 key: serialize_schema(schema)
                 for key, schema in chain(
-                    {
-                        key: getattr(self.cls, key)
-                        for key in dir(self.cls)
-                        if not key.startswith("_")
-                    }.items(),
+                    {key: getattr(self.cls, key) for key in dir(self.cls) if not key.startswith("_")}.items(),
                     typing.get_type_hints(self.cls).items(),
                 )
                 if not key.startswith("_")

sanic_openapi/oas3/__init__.py

@@ -0,0 +1,9 @@
+from collections import defaultdict
+
+from .builders import OperationBuilder, SpecificationBuilder
+
+# Static datastores, which get added to via the oas3.openapi decorators,
+# and then read from in the blueprint generation
+
+operations = defaultdict(OperationBuilder)
+specification = SpecificationBuilder()

sanic_openapi/oas3/blueprint.py

@@ -0,0 +1,142 @@
+import re
+from itertools import repeat
+from os.path import abspath, dirname, realpath
+
+from sanic.blueprints import Blueprint
+from sanic.response import json, redirect
+from sanic.views import CompositionView
+
+from ..doc import route as doc_route
+from ..utils import get_uri_filter
+from . import operations, specification
+
+
+def blueprint_factory():
+    oas3_blueprint = Blueprint("openapi", url_prefix="/swagger")
+
+    dir_path = dirname(dirname(realpath(__file__)))
+    dir_path = abspath(dir_path + "/ui")
+
+    oas3_blueprint.static("/", dir_path + "/index.html", strict_slashes=True)
+    oas3_blueprint.static("/", dir_path)
+
+    # Redirect "/swagger" to "/swagger/"
+    @oas3_blueprint.route("", strict_slashes=True)
+    def index(request):
+        return redirect("{}/".format(oas3_blueprint.url_prefix))
+
+    @oas3_blueprint.route("/swagger.json")
+    @doc_route(exclude=True)
+    def spec(request):
+        return json(specification.build().serialize())
+
+    @oas3_blueprint.route("/swagger-config")
+    def config(request):
+        return json(getattr(request.app.config, "SWAGGER_UI_CONFIGURATION", {}))
+
+    @oas3_blueprint.listener("before_server_start")
+    def build_spec(app, loop):
+        # --------------------------------------------------------------- #
+        # Globals
+        # --------------------------------------------------------------- #
+        specification.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(
+            getattr(app.config, "API_LICENSE_NAME", None),
+            getattr(app.config, "API_LICENSE_URL", None),
+        )
+
+        specification.contact(
+            getattr(app.config, "API_CONTACT_NAME", None),
+            getattr(app.config, "API_CONTACT_URL", None),
+            getattr(app.config, "API_CONTACT_EMAIL", None),
+        )
+
+        for scheme in getattr(app.config, "API_SCHEMES", ["http"]):
+            host = getattr(app.config, "API_HOST", None)
+            basePath = getattr(app.config, "API_BASEPATH", "")
+            if host is None or basePath is None:
+                continue
+
+            specification.url(f"{scheme}://{host}/{basePath}")
+
+        # --------------------------------------------------------------- #
+        # Blueprints
+        # --------------------------------------------------------------- #
+        for _blueprint in app.blueprints.values():
+            if not hasattr(_blueprint, "routes"):
+                continue
+
+            for _route in _blueprint.routes:
+                if hasattr(_route.handler, "view_class"):
+                    # class based view
+                    for http_method in _route.methods:
+                        _handler = getattr(_route.handler.view_class, http_method.lower(), None)
+                        if _handler:
+                            operation = operations[_route.handler]
+                            if not operation.tags:
+                                operation.tag(_blueprint.name)
+                else:
+                    operation = operations[_route.handler]
+                    # operation.blueprint = _blueprint
+                    # is this necc ?
+                    if not operation.tags:
+                        operation.tag(_blueprint.name)
+
+        uri_filter = get_uri_filter(app)
+
+        # --------------------------------------------------------------- #
+        # Operations
+        # --------------------------------------------------------------- #
+        for _uri, _route in app.router.routes_all.items():
+
+            # Ignore routes under swagger blueprint
+            if _route.uri.startswith(oas3_blueprint.url_prefix):
+                continue
+
+            # Apply the URI filter
+            if uri_filter(_uri):
+                continue
+
+            # route.name will be None when using class based view
+            if _route.name and "static" in _route.name:
+                continue
+
+            # --------------------------------------------------------------- #
+            # Methods
+            # --------------------------------------------------------------- #
+
+            handler_type = type(_route.handler)
+
+            if handler_type is CompositionView:
+                view = _route.handler
+                method_handlers = view.handlers.items()
+            else:
+                method_handlers = zip(_route.methods, repeat(_route.handler))
+
+            uri = _uri if _uri == "/" else _uri.rstrip("/")
+
+            for segment in _route.parameters:
+                uri = re.sub("<" + segment.name + ".*?>", "{" + segment.name + "}", uri)
+
+            for method, _handler in method_handlers:
+
+                if method == "OPTIONS":
+                    continue
+
+                operation = operations[_handler]
+
+                if not hasattr(operation, "operationId"):
+                    operation.operationId = "%s_%s" % (method.lower(), _route.name)
+
+                for _parameter in _route.parameters:
+                    operation.parameter(_parameter.name, _parameter.cast, "path")
+
+                specification.operation(uri, method, operation)
+
+    return oas3_blueprint

sanic_openapi/oas3/builders.py

@@ -0,0 +1,167 @@
+"""
+Builders for the oas3 object types
+
+These are completely internal, so can be refactored if desired without concern
+for breaking user experience
+"""
+from collections import defaultdict
+
+from .definitions import (
+    Any,
+    Contact,
+    Dict,
+    ExternalDocumentation,
+    Info,
+    License,
+    List,
+    OpenAPI,
+    Operation,
+    Parameter,
+    PathItem,
+    RequestBody,
+    Response,
+    Server,
+    Tag,
+)
+
+
+class OperationBuilder:
+    summary: str
+    description: str
+    operationId: str
+    requestBody: RequestBody
+    externalDocs: ExternalDocumentation
+    tags: List[str]
+    security: List[Any]
+    parameters: List[Parameter]
+    responses: Dict[str, Response]
+    callbacks: List[str]  # TODO
+    deprecated: bool = False
+
+    def __init__(self):
+        self.tags = []
+        self.security = []
+        self.parameters = []
+        self.responses = {}
+
+    def name(self, value: str):
+        self.operationId = value
+
+    def describe(self, summary: str = None, description: str = None):
+        if summary:
+            self.summary = summary
+
+        if description:
+            self.description = description
+
+    def document(self, url: str, description: str = None):
+        self.externalDocs = ExternalDocumentation.make(url, description)
+
+    def tag(self, *args: str):
+        for arg in args:
+            self.tags.append(arg)
+
+    def deprecate(self):
+        self.deprecated = True
+
+    def body(self, content: Any, **kwargs):
+        self.requestBody = RequestBody.make(content, **kwargs)
+
+    def parameter(self, name: str, schema: Any, location: str = "query", **kwargs):
+        self.parameters.append(Parameter.make(name, schema, location, **kwargs))
+
+    def response(self, status, content: Any = None, description: str = None, **kwargs):
+        self.responses[status] = Response.make(content, description, **kwargs)
+
+    def secured(self, *args, **kwargs):
+        items = {**{v: [] for v in args}, **kwargs}
+        gates = {}
+
+        for name, params in items.items():
+            gate = name.__name__ if isinstance(name, type) else name
+            gates[gate] = params
+
+        self.security.append(gates)
+
+    def build(self):
+        return Operation(**self.__dict__)
+
+
+class SpecificationBuilder:
+    _url: str
+    _title: str
+    _version: str
+    _description: str
+    _terms: str
+    _contact: Contact
+    _license: License
+    _paths: Dict[str, Dict[str, OperationBuilder]]
+    _tags: Dict[str, Tag]
+    # _components: ComponentsBuilder
+    # deliberately not included
+    #   -- doesnt fit in well with the auto-generated style of sanic-openapi
+    #      but could be put here down the line if desired...
+
+    def __init__(self):
+        self._paths = defaultdict(dict)
+        self._tags = {}
+
+    def url(self, value: str):
+        self._url = value
+
+    def describe(self, title: str, version: str, description: str = None, terms: str = None):
+        self._title = title
+        self._version = version
+        self._description = description
+        self._terms = terms
+
+    def tag(self, name: str, **kwargs):
+        self._tags[name] = Tag(name, **kwargs)
+
+    def contact(self, name: str = None, url: str = None, email: str = None):
+        self._contact = Contact(name=name, url=url, email=email)
+
+    def license(self, name: str = None, url: str = None):
+        self._license = License(name, url=url)
+
+    def operation(self, path: str, method: str, operation: OperationBuilder):
+        for _tag in operation.tags:
+            if _tag in self._tags.keys():
+                continue
+
+            self._tags[_tag] = Tag(_tag)
+
+        self._paths[path][method.lower()] = operation
+
+    def build(self) -> OpenAPI:
+        info = self._build_info()
+        paths = self._build_paths()
+        tags = self._build_tags()
+
+        if getattr(self, "_url", None) is not None:
+            servers = [Server(url=self._url)]
+        else:
+            servers = []
+
+        return OpenAPI(info, paths, tags=tags, servers=servers)
+
+    def _build_info(self) -> Info:
+        kwargs = {
+            "description": self._description,
+            "termsOfService": self._terms,
+            "license": self._license,
+            "contact": self._contact,
+        }
+
+        return Info(self._title, self._version, **kwargs)
+
+    def _build_tags(self):
+        return [self._tags[k] for k in self._tags]
+
+    def _build_paths(self) -> Dict:
+        paths = {}
+
+        for path, operations in self._paths.items():
+            paths[path] = PathItem(**{k: v.build() for k, v in operations.items()})
+
+        return paths