-
Notifications
You must be signed in to change notification settings - Fork 21
/
swagger_docs.py
217 lines (191 loc) · 7.89 KB
/
swagger_docs.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
import functools
import re
import warnings
from collections import defaultdict
from typing import Callable, Dict, Optional, Type, Union
import fastjsonschema
import yaml
from aiohttp import hdrs, web
from aiohttp.abc import AbstractView
from .routes import _SWAGGER_SPECIFICATION
from .swagger import ExpectHandler, Swagger
from .swagger_info import SwaggerInfo
from .swagger_route import SwaggerRoute, _SwaggerHandler
from .ui_settings import RapiDocUiSettings, ReDocUiSettings, SwaggerUiSettings
_PATH_VAR_REGEX = re.compile(r"{([_a-zA-Z][_a-zA-Z0-9].+?):.+?}(/|$)")
def swagger_doc(path: str) -> Callable:
"""
This decorator can be used if you don't want to include the whole schema into docstring,
so it can be placed in the YAML file and provide the path to the file,
see `docs_decorator_and_docstrings <https://github.com/hh-h/aiohttp-swagger3/tree/master/examples/docs_decorator_and_docstrings>`__ example
:param str path: path to swagger route schema
"""
def wrapper(fn: Callable) -> Callable:
doc = fn.__doc__ or ""
if "---" in doc:
raise Exception(f"cannot use decorator swagger_doc with docstring, function: {fn}")
with open(path) as f:
doc = f"{doc}---\n{f.read()}"
fn.__doc__ = doc
return fn
return wrapper
class SwaggerDocs(Swagger):
"""This class should be used if you want to use documentation through handler doctrings.
:param aiohttp.web.Application app: aiohttp's Application instance
:param bool validate: if ``False``, request validation is disabled, default ``True``
:param str request_key: key name under which the data will be stored in ``request``
after validation, default ``data``
:param str title: title which will be used in openapi scheme, default ``OpenAPI3``
:param str version: version which will be used in openapi scheme, default ``1.0.0``
:param str description: description which will be used in openapi scheme (optional)
:param str components: path to file with components definitions (optional)
:param str security: path to file with security definition (optional)
:param swagger_ui_settings: class:`SwaggerUiSettings` (optional)
:param redoc_ui_settings: class:`ReDocUiSettings` (optional)
:param rapidoc_ui_settings: class:`RapiDocUiSettings` (optional)
"""
__slots__ = ()
def __init__(
self,
app: web.Application,
*,
validate: bool = True,
info: Optional[SwaggerInfo] = None,
request_key: str = "data",
title: Optional[str] = None,
version: Optional[str] = None,
description: Optional[str] = None,
components: Optional[str] = None,
security: Optional[str] = None,
swagger_ui_settings: Optional[SwaggerUiSettings] = None,
redoc_ui_settings: Optional[ReDocUiSettings] = None,
rapidoc_ui_settings: Optional[RapiDocUiSettings] = None,
) -> None:
if info is not None and (title is not None or version is not None or description is not None):
raise Exception("do not use SwaggerDocs' info with title or version or description")
if info is None:
info = SwaggerInfo(title="OpenAPI3", version="1.0.0")
if title is not None:
warnings.warn(
"SwaggerDocs' title is deprecated and will be removed in 0.8.0, use info object instead.",
FutureWarning,
)
info.title = title
if version is not None:
warnings.warn(
"SwaggerDocs' version is deprecated and will be removed in 0.8.0, use info object instead.",
FutureWarning,
)
info.version = version
if description is not None:
warnings.warn(
"SwaggerDocs' description is deprecated and will be removed in 0.8.0, use info object instead.",
FutureWarning,
)
info.description = description
spec: Dict = {
"openapi": "3.0.0",
"info": info.to_json(),
"paths": defaultdict(lambda: defaultdict(dict)),
}
if components:
with open(components) as f:
spec.update(yaml.safe_load(f))
if security:
with open(security) as f:
spec.update(yaml.safe_load(f))
super().__init__(
app,
validate=validate,
spec=spec,
request_key=request_key,
swagger_ui_settings=swagger_ui_settings,
redoc_ui_settings=redoc_ui_settings,
rapidoc_ui_settings=rapidoc_ui_settings,
)
self._app[_SWAGGER_SPECIFICATION] = self.spec
def _wrap_handler(
self,
method: str,
path: str,
handler: _SwaggerHandler,
*,
is_method: bool,
validate: bool,
) -> _SwaggerHandler:
if not handler.__doc__ or "---" not in handler.__doc__:
return handler
*_, spec = handler.__doc__.split("---")
method_spec = yaml.safe_load(spec)
path = _PATH_VAR_REGEX.sub(r"{\1}\2", path)
if self.spec["paths"].get(path, {}).get(method) is not None:
raise Exception(f"{method} {path} already exists")
self.spec["paths"][path][method] = method_spec
try:
self.spec_validate(self.spec)
except fastjsonschema.exceptions.JsonSchemaException as exc:
fn_name = handler.__name__
raise Exception(f"Invalid schema for handler '{fn_name}' {method.upper()} {path} - {exc}")
self._app[_SWAGGER_SPECIFICATION] = self.spec
if not validate:
return handler
route = SwaggerRoute(method, path, handler, swagger=self)
if is_method:
return functools.partialmethod(self._handle_swagger_method_call, route) # type: ignore
return functools.partial(self._handle_swagger_call, route)
def add_route(
self,
method: str,
path: str,
handler: Union[_SwaggerHandler, Type[AbstractView]],
*,
name: Optional[str] = None,
expect_handler: Optional[ExpectHandler] = None,
validate: Optional[bool] = None,
) -> web.AbstractRoute:
if validate is None:
need_validation: bool = self.validate
else:
need_validation = False if not self.validate else validate
if isinstance(handler, type) and issubclass(handler, AbstractView):
for meth in hdrs.METH_ALL:
meth = meth.lower()
handler_ = getattr(handler, meth, None)
if handler_ is not None:
setattr(
handler,
meth,
self._wrap_handler(
meth,
path,
handler_,
is_method=True,
validate=need_validation,
),
)
else:
if method == hdrs.METH_ANY:
for meth in (
hdrs.METH_GET,
hdrs.METH_POST,
hdrs.METH_PUT,
hdrs.METH_PATCH,
hdrs.METH_DELETE,
):
meth = meth.lower()
handler = self._wrap_handler(
meth,
path,
handler,
is_method=False,
validate=need_validation,
)
else:
handler = self._wrap_handler(
method.lower(),
path,
handler,
is_method=False,
validate=need_validation,
)
return self._app.router.add_route(method, path, handler, name=name, expect_handler=expect_handler)