Revamp OAS implementation

sanic_ext/__init__.py

@@ -4,6 +4,7 @@
 from sanic_ext.config import Config
 from sanic_ext.extensions.base import Extension
 from sanic_ext.extensions.http.cors import cors
+from sanic_ext.extensions.oas import expose as oas_experimental
 from sanic_ext.extensions.openapi import openapi
 from sanic_ext.extensions.templating.render import render
 from sanic_ext.extras.request import CountedRequest
@@ -22,4 +23,5 @@
     "render",
     "serializer",
     "validate",
+    "oas_experimental",
 ]

sanic_ext/bootstrap.py

@@ -20,6 +20,7 @@
     InjectionRegistry,
 )
 from sanic_ext.extensions.logging.extension import LoggingExtension
+from sanic_ext.extensions.oas.extension import OASExtension
 from sanic_ext.extensions.openapi.builders import SpecificationBuilder
 from sanic_ext.extensions.openapi.extension import OpenAPIExtension
 from sanic_ext.utils.string import camel_to_snake
@@ -92,8 +93,9 @@ def __init__(
             extensions.extend(
                 [
                     InjectionExtension,
-                    OpenAPIExtension,
-                    HTTPExtension,
+                    # OpenAPIExtension,
+                    OASExtension,
+                    # HTTPExtension,
                     HealthExtension,
                     LoggingExtension,
                 ]

sanic_ext/extensions/oas/__init__.py

@@ -0,0 +1 @@
+from .expose import *  # noqa

sanic_ext/extensions/oas/base.py

@@ -0,0 +1,26 @@
+from abc import ABC
+from functools import partial, wraps
+
+
+class BaseDecorator(ABC):
+    def __get__(self, instance, _):
+        return wraps(self._func)(partial(self.__call__, instance))
+
+    def __call__(self, func):
+        self._func = func
+        self.setup()
+
+        def decorator(f):
+            @wraps(f)
+            def decorated_function(*a, **kw):
+                return self.execute(a, kw)
+
+            return decorated_function
+
+        return decorator(func)
+
+    def setup(self):
+        ...
+
+    def execute(self, args, kwargs):
+        return self._func(*args, **kwargs)

sanic_ext/extensions/oas/builder.py

@@ -0,0 +1,33 @@
+from typing import Optional
+
+from sanic import Sanic  # type: ignore
+
+from .registry import DefinitionRegistry
+from .schema import OpenAPI, Paths
+
+
+class OASBuilder:
+    def __init__(self, app: Sanic, paths: Paths):
+        self._app = app
+        self._registry = DefinitionRegistry()
+        self._paths = paths
+        self._oas: Optional[OpenAPI] = None
+
+    @property
+    def oas(self) -> OpenAPI:
+        if not self._oas:
+            self._oas = OpenAPI()
+        return self._oas
+
+    def build(self):
+        self._oas = OpenAPI(
+            info={
+                "title": self._app.name,
+                "version": "0.0.0",
+            },
+            paths=self._paths,
+        )
+
+        from rich import print
+
+        print(self.oas.serialize())

sanic_ext/extensions/oas/decorators/base.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+from abc import ABCMeta
+from enum import Enum, auto
+from types import SimpleNamespace
+from typing import Any, Dict
+
+from ..base import BaseDecorator
+from ..registry import DefinitionRegistry
+
+DEFAULT_META = {
+    "duplicate": False,
+    "merge": False,
+}
+
+
+class BaseEnum(str, Enum):
+    def _generate_next_value_(name: str, *args) -> str:  # type: ignore
+        return name.lower()
+
+    def __str__(self):
+        return self.name.lower()
+
+
+class ParameterInChoice(BaseEnum):
+    QUERY = auto()
+    HEADER = auto()
+    PATH = auto()
+    COOKIE = auto()
+
+
+class ParameterStyleChoice(BaseEnum):
+    DEFAULT = auto()
+    FORM = auto()
+    SIMPLE = auto()
+    MATRIX = auto()
+    LABEL = auto()
+    SPACE_DELIMITED = auto()
+    PIPE_DELIMITED = auto()
+    DEEP_OBJECT = auto()
+
+
+class DefinitionType(ABCMeta):
+    def __new__(cls, name, bases, attrs, **kwargs):
+        meta = {**DEFAULT_META}
+        if defined := attrs.pop("Meta", None):
+            meta.update(
+                {
+                    k: v
+                    for k, v in defined.__dict__.items()
+                    if not k.startswith("_")
+                }
+            )
+        attrs["meta"] = SimpleNamespace(**meta)
+        gen_class = super().__new__(cls, name, bases, attrs, **kwargs)
+        return gen_class
+
+
+class Definition(BaseDecorator, metaclass=DefinitionType):
+    def setup(self):
+        key = f"{self._func.__module__}.{self._func.__qualname__}"
+        DefinitionRegistry()[key] = self
+        super().setup()
+
+
+class Extendable:
+    """
+    This object is extendable according to Specification Extensions. For more
+    information: https://swagger.io/specification/#specification-extensions
+
+    As an example:
+
+        @oas.foo(..., something_additional=1234)
+
+    Will translate into the property:
+
+        x-something-additional
+    """
+
+    extension: Dict[str, Any]
+
+    def __new__(cls, *args, **kwargs):
+        if args and args[0] is cls:
+            args = args[1:]
+        if not hasattr(cls, "_init"):
+            cls._init = cls.__init__
+            cls.__init__ = lambda *a, **k: None
+
+        obj = object.__new__(cls)
+        obj.extension = {
+            key: kwargs.pop(key)
+            for key in list(kwargs.keys())
+            if key not in cls.__dataclass_fields__
+        }
+        cls._init(obj, *args, **kwargs)
+
+        return obj

sanic_ext/extensions/oas/decorators/objects.py

@@ -0,0 +1,194 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Type, Union, get_type_hints
+
+from ..definition import Definition
+from ..signature import (
+    Deprecated,
+    ExternalDocument,
+    Parameter,
+    ParameterDict,
+    Reference,
+    RequestBody,
+    Response,
+    ResponseReference,
+    SecurityScheme,
+    Text,
+)
+
+
+@dataclass
+class tags(Definition):
+    values: List[str]
+
+    class Meta:
+        merge = True
+
+    def merge(self, target: tags) -> None:
+        self.values.extend(tags.values)
+
+    def serialize(self) -> Dict[str, Any]:
+        return {"tags": self.values}
+
+
+@dataclass
+class tag(Text, Definition):
+    def setup(self):
+        tags([self.text])(self._func)
+
+
+@dataclass
+class summary(Text, Definition):
+    ...
+
+
+@dataclass
+class description(Text, Definition):
+    ...
+
+
+@dataclass
+class external_docs(Definition):
+    documents: List[ExternalDocument]
+
+    class Meta:
+        merge = True
+
+    def merge(self, target: external_docs) -> None:
+        self.documents.extend(target.documents)
+
+
+@dataclass
+class external_doc(ExternalDocument, Definition):
+    def setup(self):
+        external_docs([self])(self._func)
+
+
+document = external_doc
+
+
+@dataclass
+class operation_id(Text, Definition):
+    ...
+
+
+@dataclass
+class parameters(Definition):  # TODO: Reference
+    values: List[Union[Parameter, Reference, ParameterDict]] = field(
+        default_factory=list
+    )
+    model: Optional[Type] = None
+
+    class Meta:
+        merge = True
+
+    def merge(self, target: parameters) -> None:
+        names = [
+            name for val in self.values if (name := getattr(val, "name", None))
+        ]
+        self.values.extend(
+            [
+                Parameter(**value) if isinstance(value, dict) else value
+                for value in target.values
+                if (
+                    (isinstance(value, Parameter) and value.name not in names)
+                    or (isinstance(value, dict) and value["name"] not in names)
+                    # TODO: dedupe references
+                    or isinstance(value, Reference)
+                )
+            ]
+        )
+
+    def __post_init__(self):
+        self.values = [self.parameterize(val) for val in self.values]
+        if self.model:
+            self.values.extend(
+                [
+                    Parameter(name=name, schema=schema)
+                    for name, schema in get_type_hints(self.model).items()
+                ]
+            )
+
+    def serialize(self) -> Dict[str, Any]:
+        return {
+            "parameters": [
+                val if isinstance(val, dict) else val.serialize()
+                for val in self.values
+            ]
+        }
+
+    @staticmethod
+    def parameterize(item: Any) -> Parameter:
+        if isinstance(item, Parameter):
+            return item
+        elif isinstance(item, dict):
+            return Parameter(**item)
+        # TODO: Reference
+        else:
+            raise TypeError(
+                f"Expected Parameter or dict, got {type(item).__name__}"
+            )
+
+
+@dataclass
+class parameter(Parameter, Definition):
+    def setup(self):
+        parameters([self])(self._func)
+
+
+@dataclass
+class request_body(RequestBody, Reference, Definition):  # TODO
+    ...
+
+
+body = request_body
+
+
+@dataclass
+class responses(Definition):
+    values: List[Union[Response, ResponseReference]]
+
+    class Meta:
+        merge = True
+
+    def merge(self, target: responses) -> None:
+        self.values.extend(target.values)
+
+    def serialize(self) -> Dict[str, Any]:
+        return {
+            "responses": {
+                "default"
+                if isinstance(val, dict)
+                else val.status: val
+                if isinstance(val, dict)
+                else val.serialize()
+                for val in self.values
+            }
+        }
+
+
+@dataclass
+class response(Response, Definition):
+    def setup(self):
+        responses(values=[self])(self._func)
+
+
+@dataclass
+class callbacks(Definition):  # TODO
+    ...
+
+
+@dataclass
+class deprecated(Deprecated, Definition):
+    ...
+
+
+@dataclass
+class security(SecurityScheme, Definition):
+    ...
+
+
+@dataclass
+class servers(Definition):  # TODO
+    ...