diff --git a/sanic_ext/__init__.py b/sanic_ext/__init__.py index b9ba15d..36f8148 100644 --- a/sanic_ext/__init__.py +++ b/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", ] diff --git a/sanic_ext/bootstrap.py b/sanic_ext/bootstrap.py index bf149ca..42c8e27 100644 --- a/sanic_ext/bootstrap.py +++ b/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, ] diff --git a/sanic_ext/extensions/oas/__init__.py b/sanic_ext/extensions/oas/__init__.py new file mode 100644 index 0000000..0286fe6 --- /dev/null +++ b/sanic_ext/extensions/oas/__init__.py @@ -0,0 +1 @@ +from .expose import * # noqa diff --git a/sanic_ext/extensions/oas/base.py b/sanic_ext/extensions/oas/base.py new file mode 100644 index 0000000..463aea6 --- /dev/null +++ b/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) diff --git a/sanic_ext/extensions/oas/builder.py b/sanic_ext/extensions/oas/builder.py new file mode 100644 index 0000000..035eaa6 --- /dev/null +++ b/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()) diff --git a/sanic_ext/extensions/oas/decorators/__init__.py b/sanic_ext/extensions/oas/decorators/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/sanic_ext/extensions/oas/decorators/base.py b/sanic_ext/extensions/oas/decorators/base.py new file mode 100644 index 0000000..a97f1f9 --- /dev/null +++ b/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 diff --git a/sanic_ext/extensions/oas/decorators/objects.py b/sanic_ext/extensions/oas/decorators/objects.py new file mode 100644 index 0000000..986dc28 --- /dev/null +++ b/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 + ... diff --git a/sanic_ext/extensions/oas/definition.py b/sanic_ext/extensions/oas/definition.py new file mode 100644 index 0000000..212333c --- /dev/null +++ b/sanic_ext/extensions/oas/definition.py @@ -0,0 +1,136 @@ +from __future__ import annotations + +from abc import ABCMeta +from enum import Enum, auto +from types import SimpleNamespace +from typing import Any, Dict + +from sanic.helpers import Default + +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() + + def serialize(self) -> str: + return str(self) + + +class ParameterInChoice(BaseEnum): + QUERY = auto() + HEADER = auto() + PATH = auto() + COOKIE = auto() + + +class ParameterStyleChoice(BaseEnum): + 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 Serializable: + def serialize(self) -> Dict[str, Any]: + return { + self.serialize_key(key): self.serialize_value(value) + for key, value in self.__dict__.items() + if (not key.startswith("_") and not isinstance(value, Default)) + } + + @staticmethod + def serialize_key(key: str) -> str: + key = "".join( + word.capitalize() if i else word + for i, word in enumerate(key.rstrip("_").split("_")) + ) + return key + + @staticmethod + def serialize_value(value: Any) -> Any: + if hasattr(value, "serialize"): + return value.serialize() + elif isinstance(value, list): + return [Serializable.serialize_value(i) for i in value] + elif isinstance(value, dict): + return { + key: Serializable.serialize_value(value) + for key, value in value.items() + } + else: + return value + + +class Definition(BaseDecorator, Serializable, metaclass=DefinitionType): + def setup(self): + key = f"{self._func.__module__}.{self._func.__qualname__}" + DefinitionRegistry()[key] = self + + +class Extendable(Serializable): + """ + 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 + + def serialize(self) -> Dict[str, Any]: + base = super().serialize() + base.update(self._extension) + return base diff --git a/sanic_ext/extensions/oas/expose.py b/sanic_ext/extensions/oas/expose.py new file mode 100644 index 0000000..5f1e826 --- /dev/null +++ b/sanic_ext/extensions/oas/expose.py @@ -0,0 +1,43 @@ +from .decorators.objects import ( + body, + callbacks, + deprecated, + description, + document, + external_doc, + external_docs, + operation_id, + parameter, + parameters, + request_body, + response, + responses, + security, + servers, + summary, + tag, + tags, +) +from .signature import Schema + +__all__ = ( + "Schema", + "body", + "callbacks", + "deprecated", + "description", + "document", + "external_doc", + "external_docs", + "operation_id", + "parameters", + "request_body", + "response", + "responses", + "security", + "servers", + "summary", + "tag", + "tags", + "parameter", +) diff --git a/sanic_ext/extensions/oas/extension.py b/sanic_ext/extensions/oas/extension.py new file mode 100644 index 0000000..39265b4 --- /dev/null +++ b/sanic_ext/extensions/oas/extension.py @@ -0,0 +1,11 @@ +from ..base import Extension +from .startup import build_spec + + +class OASExtension(Extension): + name = "oas" + + def startup(self, bootstrap) -> None: + if self.app.config.OAS: + bootstrap.oas = None + build_spec(self.app) diff --git a/sanic_ext/extensions/oas/registry.py b/sanic_ext/extensions/oas/registry.py new file mode 100644 index 0000000..a8dee34 --- /dev/null +++ b/sanic_ext/extensions/oas/registry.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, List, Type, TypeVar + +if TYPE_CHECKING: + from .decorators import Definition # type: ignore +else: + Definition = TypeVar("Operation") + + +class DefinitionRegistry(dict[str, List[Definition]]): + _registry: DefinitionRegistry + + def __setitem__(self, key: str, definition: Definition) -> None: + self.setdefault(key, list()) + existing = next( + (val for val in self[key] if isinstance(val, type(definition))), + None, + ) + if existing: + if definition.meta.merge: + existing.merge(definition) + return + if not definition.meta.duplicate: + raise RuntimeError( + f"Cannot define multiple {definition.__class__.__name__} " + f"on {key}" + ) + self[key].append(definition) + + def __new__( + cls: Type[DefinitionRegistry], *args, **kwargs + ) -> DefinitionRegistry: + if existing := getattr(cls, "_registry", None): + return existing + cls._registry = super().__new__(cls) # type: ignore + return cls._registry diff --git a/sanic_ext/extensions/oas/schema.py b/sanic_ext/extensions/oas/schema.py new file mode 100644 index 0000000..d6842f8 --- /dev/null +++ b/sanic_ext/extensions/oas/schema.py @@ -0,0 +1,116 @@ +import re +from dataclasses import dataclass, field +from functools import cached_property +from typing import Any, List + +from sanic.constants import HTTPMethod +from sanic_routing.route import Route + +from .decorators.objects import parameters +from .definition import Definition, Serializable +from .signature import Parameter, ParameterInChoice, SecurityScheme + + +@dataclass +class Path: + route: Route + method: HTTPMethod + definitions: List[Definition] = field(default_factory=list) + _PATTERN = re.compile( + r"<(?P[a-zA-Z0-9_]+)(?::(?P[a-zA-Z0-9_]+))?>" + ) + + def __post_init__(self): + parameters( + values=[ + Parameter( + name=param.name, + in_=ParameterInChoice.PATH, + schema=param.cast, + ) + for param in self.route.defined_params.values() + ] + )(self.route.handler) + + @cached_property + def uri(self) -> str: + return self._convert_uri(self.route.uri) + + def _convert_uri(self, uri: str) -> str: + return self._PATTERN.sub(r"{\g}", uri) + + +@dataclass +class Paths: + _items: List[Path] = field(default_factory=list) + + def register(self, item: Path) -> None: + self._items.append(item) + + def serialize(self) -> dict[str, Any]: + output: dict[str, Any] = {} + + for item in self._items: + uri = item.uri + path_params = [ + param.name for param in item.route.defined_params.values() + ] + output.setdefault(uri, {}) + for definition in item.definitions: + output[uri].setdefault(item.method.lower(), {}) + output[uri][item.method.lower()].update(definition.serialize()) + + # Since OAS definitions need to be added using the handler, + # when multiple route definitions are on a single handler it + # is possible for path params to be defined on the handler + # that are not in the route definition. Therefore we need to + # remove any path params that are not defined in the route. + if "parameters" in output[uri][item.method.lower()]: + output[uri][item.method.lower()]["parameters"] = [ + param + for param in output[uri][item.method.lower()][ + "parameters" + ] + if param["in"] != "path" + or param["name"] in path_params + ] + + if "responses" not in output[uri][item.method.lower()]: + output[uri][item.method.lower()]["responses"] = { + "default": {"description": "Default response"} + } + + return output + + +@dataclass +class Components: + schemas: dict[str, Any] = field(default_factory=dict) + security_schemes: list[dict[str, Any]] = field(default_factory=list) + parameters: dict[str, Any] = field(default_factory=dict) + responses: dict[str, Any] = field(default_factory=dict) + requestBodies: dict[str, Any] = field(default_factory=dict) + headers: dict[str, Any] = field(default_factory=dict) + examples: dict[str, Any] = field(default_factory=dict) + links: dict[str, Any] = field(default_factory=dict) + callbacks: dict[str, Any] = field(default_factory=dict) + path_items: dict[str, Any] = field(default_factory=dict) + + def serialize(self) -> dict[str, Any]: + output: dict[str, Any] = {} + for key, value in self.__dict__.items(): + if value: + output[key] = ( + value.serialize() + if isinstance(value, Serializable) + else value + ) + return output + + +@dataclass +class OpenAPI(Serializable): + openapi: str = "3.0.3" + info: dict[str, Any] = field(default_factory=dict) + paths: Paths = field(default_factory=Paths) + components: Components = field(default_factory=Components) diff --git a/sanic_ext/extensions/oas/signature.py b/sanic_ext/extensions/oas/signature.py new file mode 100644 index 0000000..9faaca8 --- /dev/null +++ b/sanic_ext/extensions/oas/signature.py @@ -0,0 +1,462 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date, datetime, timedelta +from typing import Any, Dict, List, Literal, Optional, Type, TypedDict, Union +from uuid import UUID + +from sanic.helpers import Default, _default + +from .definition import ( + Extendable, + ParameterInChoice, + ParameterStyleChoice, + Serializable, +) + +try: + from typing import NotRequired # type: ignore +except ImportError: + from typing_extensions import NotRequired + + +@dataclass +class Description(Serializable): + description: Union[str, Default] = _default + + +@dataclass +class URL(Serializable): + url: str + + +@dataclass +class ExternalDocument(Description, URL, Extendable): + """Some documentation about external documents""" + + +@dataclass +class Name(Serializable): + name: str + + +@dataclass +class Text(Serializable): + text: str + + +@dataclass +class Summary(Serializable): + summary: Union[str, Default] = _default + + +@dataclass +class Required(Serializable): + required: Union[bool, Default] = _default + + +@dataclass +class Deprecated(Serializable): + deprecated: bool = True + + +@dataclass +class ParameterIn(Serializable): + in_: ParameterInChoice = ParameterInChoice.QUERY + + +@dataclass +class Schema( + Serializable +): # TODO: Schema https://swagger.io/specification/#schema-object + model: Type[object] = str + format: Union[ + Literal["date-time"], + Literal["date"], + Literal["time"], + Literal["duration"], + Literal["email"], + Literal["idn-email"], + Literal["hostname"], + Literal["idn-hostname"], + Literal["ipv4"], + Literal["ipv6"], + Literal["uuid"], + Literal["uri"], + Literal["uri-reference"], + Literal["iri"], + Literal["iri-reference"], + Literal["uri-template"], + Literal["json-pointer"], + Literal["relative-json-pointer"], + Literal["regex"], + Default, + ] = _default + + def __post_init__(self): + if isinstance(self.model, Default): + self.model = str + elif self.model is UUID: + self.model = str + self.format = "uuid" + elif self.model is datetime: + self.model = str + self.format = "date-time" + elif self.model is date: + self.model = str + self.format = "date" + elif self.model is timedelta: + self.model = str + self.format = "duration" + + self._format_validation() + + def _format_validation(self): + # TODO: https://swagger.io/specification/#data-types + if isinstance(self.model, Default) or isinstance(self.format, Default): + return + + if ( + ( + self.model is int + and self.format + in ( + "int32", + "int64", + ) + ) + or ( + self.model is float + and self.format + in ( + "float", + "double", + ) + ) + or ( + self.model is str + and self.format + in ( + "date-time", + "date", + "time", + "duration", + "email", + "idn-email", + "hostname", + "idn-hostname", + "ipv4", + "ipv6", + "uuid", + "uri", + "uri-reference", + "iri", + "iri-reference", + "uri-template", + "json-pointer", + "relative-json-pointer", + "regex", + ) + ) + ): + return + + raise ValueError(f"Invalid format for {self.model}: {self.format}") + + def serialize(self) -> Dict[str, Any]: + base = super().serialize() + model = base.pop("model") + + # TODO: https://swagger.io/specification/#data-types + if model is str: + base["type"] = "string" + elif model is int: + base["type"] = "integer" + base.setdefault("format", "int32") + elif model is float: + base["type"] = "number" + base["format"] = "float" + elif model is bool: + base["type"] = "boolean" + elif model is list: + base["type"] = "array" + elif model is dict: + base["type"] = "object" + else: + # TODO: Create a component for this model + # https://swagger.io/specification/#reference-object + # base["$ref"] = f"#/components/schemas/{self.model.__name__}" + ... + + return base + + +@dataclass +class SchemaArg(Serializable): + schema: Optional[ + Union[Default, Schema, str, float, int, bool, Reference] + ] = _default + + def __post_init__(self): + if isinstance(self.schema, dict): + self.schema = Schema(**self.schema) + + +@dataclass +class SimpleExample(Serializable): + example: Union[Any, Default] = _default + + +@dataclass +class Example(Description, Summary, Extendable): + value: Any = None + external_value: Union[Default, str] = _default + + def __post_init__(self): + if not isinstance(self.external_value, Default) and self.value: + raise ValueError + super().__post_init__() + + +@dataclass +class Examples(Serializable): + examples: Union[ + List[Dict[str, Union[Example, Reference]]], Default + ] = _default + + +@dataclass +class Content(Serializable): + content: Optional[ + Union[Default, Dict[str, Union[MediaType, MediaTypeDict]]] + ] = _default + + def __post_init__(self): + if isinstance(self.content, dict): + self.content = { + key: MediaType(**value) for key, value in self.content.items() + } + + +@dataclass +class Parameter( + Examples, + SimpleExample, + Deprecated, + Required, + Description, + Content, + SchemaArg, + ParameterIn, + Name, + Extendable, +): + style: Union[ParameterStyleChoice, Default] = _default + allow_empty_value: Union[Default, bool] = _default + allow_reserved: Union[Default, bool] = _default + explode: Union[Default, bool] = _default + + __ENUMS__ = (("in_", ParameterInChoice), ("style", ParameterStyleChoice)) + + def __post_init__(self): + for name, choices in self.__ENUMS__: + value = getattr(self, name) + if not isinstance(value, choices) and not isinstance( + value, Default + ): + setattr(self, name, choices(value)) + super().__post_init__() + + if self.in_ is ParameterInChoice.PATH: + self.required = True + + # if self.style is ParameterStyleChoice.DEFAULT: + # self.style = ( + # ParameterStyleChoice.FORM + # if self.in_ + # in (ParameterInChoice.QUERY, ParameterInChoice.COOKIE) + # else ParameterStyleChoice.SIMPLE + # ) + + if self.allow_empty_value and self.in_ is not ParameterInChoice.QUERY: + raise ValueError + + # if isinstance(self.explode, Default): + # self.explode = self.style is ParameterStyleChoice.FORM + + if self.example and self.examples: + raise ValueError("Cannot have both example and examples") + # elif self.example: + # self.examples = [Example(value=self.example)] + # self.example = _default + + if isinstance(self.schema, Default) and isinstance( + self.content, Default + ): + self.schema = str + # self.content = None + # elif isinstance(self.schema, Default) and not isinstance( + # self.content, Default + # ): + # self.schema = None + # elif not isinstance(self.schema, Default) and isinstance( + # self.content, Default + # ): + # self.content = None + # else: + elif not isinstance(self.schema, Default) and not isinstance( + self.content, Default + ): + raise ValueError("Cannot have both schema and content") + + if self.schema and not isinstance(self.schema, Schema): + self.schema = Schema(model=self.schema) + + +@dataclass +class Reference(Serializable): + ... + + +@dataclass +class Header( + Examples, + SimpleExample, + Deprecated, + Required, + Description, + Content, + SchemaArg, + Extendable, +): + allow_empty_value: Union[Default, bool] = _default + allow_reserved: Union[Default, bool] = _default + explode: Union[Default, bool] = _default + + +@dataclass +class Encoding(Serializable): + content_type: Union[str, Default] = _default + headers: Union[Dict[str, Union[Header, Reference]], Default] = _default + style: Union[str, Default] = _default + explode: Union[bool, Default] = _default + allow_reserved: Union[bool, Default] = _default + + +@dataclass +class MediaType(Examples, SimpleExample, SchemaArg, Encoding, Extendable): + def __post_init__(self): + if isinstance(self.schema, Default): + raise ValueError("MediaType must have a schema") + if not isinstance(self.schema, Schema): + self.schema = Schema(model=self.schema) + super().__post_init__() + + +@dataclass +class Link(Description): + ... + # TODO: https://swagger.io/specification/#link-object + + +@dataclass +class RequestBody(Required, Description, Content): + def __post_init__(self): + if isinstance(self.content, Default): + raise ValueError + + +@dataclass +class Headers(Serializable): + headers: Union[Dict[str, Union[Header, Reference]], Default] = _default + + +@dataclass +class Links(Serializable): + links: Union[Dict[str, Union[Link, Reference]], Default] = _default + + +@dataclass +class Status(Serializable): + status_code: Union[int, Default] = _default + + def serialize(self) -> Dict[str, Any]: + base = super().serialize() + base.pop("statusCode", None) + return base + + @property + def status(self) -> str: + if isinstance(self.status_code, Default): + return "default" + return str(self.status_code) + + +# TODO: +# - headers +# - links +@dataclass +class Response(Description, Headers, Content, Links, Status): + def __post_init__(self): + if isinstance(self.content, Default): + raise ValueError("Response must have content") + if not self.description or isinstance(self.description, Default): + self.description = "" + super().__post_init__() + + +@dataclass +class ResponseReference(Reference, Status): + ... + + +@dataclass +class Callback(Serializable): + ... + + +@dataclass +class SecurityScheme(Extendable): + scheme: Union[Dict[str, List[str]], Default] = _default + schemes: Union[List[Dict[str, List[str]]], Default] = _default + + def __post_init__(self): + if isinstance(self.schemes, Default) and isinstance( + self.scheme, Default + ): + raise ValueError("SecurityScheme must have a scheme") + if not isinstance(self.schemes, Default) and not isinstance( + self.scheme, Default + ): + raise ValueError("Cannot have both scheme and schemes") + if not isinstance(self.scheme, Default): + self.schemes = [self.scheme] + self.scheme = _default + + def serialize(self) -> Dict[str, Any]: + return {"security": self.schemes} + + +# !!!!!!! DICTS !!!!!!! # + + +class ParameterDict(TypedDict): + name: str + in_: ParameterInChoice + description: NotRequired[str] + required: NotRequired[bool] + deprecated: NotRequired[bool] + allow_empty_value: NotRequired[bool] + allow_reserved: NotRequired[bool] + style: NotRequired[ParameterStyleChoice] + explode: NotRequired[bool] + schema: NotRequired[Union[Schema, str, float, int, bool, Reference]] + example: NotRequired[Any] + examples: NotRequired[List[Dict[str, Union[Example, Reference]]]] + + +class MediaTypeDict(TypedDict): + schema: Union[Schema, str, float, int, bool, Reference] + example: NotRequired[Any] + examples: NotRequired[List[Dict[str, Union[Example, Reference]]]] + encoding: NotRequired[Dict[str, Union[Encoding, Reference]]] diff --git a/sanic_ext/extensions/oas/startup.py b/sanic_ext/extensions/oas/startup.py new file mode 100644 index 0000000..3ff16d6 --- /dev/null +++ b/sanic_ext/extensions/oas/startup.py @@ -0,0 +1,32 @@ +from sanic import Sanic # type: ignore + +from sanic_ext.extensions.oas.builder import OASBuilder +from sanic_ext.extensions.oas.registry import DefinitionRegistry + +from .schema import Path, Paths + + +async def _build(app: Sanic): + registry = DefinitionRegistry() + + paths = Paths() + + for route in app.router.routes: + wrapped = getattr(route.handler, "__wrapped__", None) + if not wrapped and getattr(route.handler, "view_class", None): + if len(route.methods) > 1: + raise Exception + wrapped = getattr( + route.handler.view_class, tuple(route.methods)[0].lower(), None + ) + if wrapped: + key = f"{wrapped.__module__}.{wrapped.__qualname__}" + if definition := registry.get(key): + for method in route.methods: + paths.register(Path(route, method, definition)) + app.ctx._oas_builder = OASBuilder(app, paths) + app.ctx._oas_builder.build() + + +def build_spec(app: Sanic): + app.after_server_start(_build)