Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/connexion/middleware/swagger_ui.py: 42%

1import json 

2import logging 

3import pathlib 

4import re 

5import typing as t 

6from contextvars import ContextVar 

7 

8from starlette.requests import Request as StarletteRequest 

9from starlette.responses import RedirectResponse 

10from starlette.responses import Response as StarletteResponse 

11from starlette.routing import Router 

12from starlette.staticfiles import StaticFiles 

13from starlette.templating import Jinja2Templates 

14from starlette.types import ASGIApp, Receive, Scope, Send 

15 

16from connexion.jsonifier import Jsonifier 

17from connexion.middleware import SpecMiddleware 

18from connexion.middleware.abstract import AbstractSpecAPI 

19from connexion.options import SwaggerUIOptions 

20from connexion.utils import yamldumper 

21 

22logger = logging.getLogger("connexion.middleware.swagger_ui") 

23 

24 

25_original_scope: ContextVar[Scope] = ContextVar("SCOPE") 

26 

27 

28class SwaggerUIAPI(AbstractSpecAPI): 

29 def __init__( 

30 self, 

31 *args, 

32 default: ASGIApp, 

33 swagger_ui_options: t.Optional[dict] = None, 

34 **kwargs 

35 ): 

36 super().__init__(*args, **kwargs) 

37 

38 self.router = Router(default=default) 

39 self.options = SwaggerUIOptions( 

40 swagger_ui_options, oas_version=self.specification.version 

41 ) 

42 

43 if self.options.openapi_spec_available: 

44 self.add_openapi_json() 

45 self.add_openapi_yaml() 

46 

47 if self.options.openapi_console_ui_available: 

48 self.add_swagger_ui() 

49 

50 self._templates = Jinja2Templates( 

51 directory=str(self.options.openapi_console_ui_from_dir) 

52 ) 

53 

54 @staticmethod 

55 def normalize_string(string): 

56 return re.sub(r"[^a-zA-Z0-9]", "_", string.strip("/")) 

57 

58 def _base_path_for_prefix(self, request: StarletteRequest) -> str: 

59 """ 

60 returns a modified basePath which includes the incoming root_path. 

61 """ 

62 return request.scope.get("root_path", "").rstrip("/") 

63 

64 def _spec_for_prefix(self, request): 

65 """ 

66 returns a spec with a modified basePath / servers block 

67 which corresponds to the incoming request path. 

68 This is needed when behind a path-altering reverse proxy. 

69 """ 

70 base_path = self._base_path_for_prefix(request) 

71 return self.specification.with_base_path(base_path).raw 

72 

73 def add_openapi_json(self): 

74 """ 

75 Adds openapi json to {base_path}/openapi.json 

76 (or {base_path}/swagger.json for swagger2) 

77 """ 

78 logger.info( 

79 "Adding spec json: %s%s", self.base_path, self.options.openapi_spec_path 

80 ) 

81 self.router.add_route( 

82 methods=["GET"], 

83 path=self.options.openapi_spec_path, 

84 endpoint=self._get_openapi_json, 

85 ) 

86 

87 def add_openapi_yaml(self): 

88 """ 

89 Adds openapi json to {base_path}/openapi.json 

90 (or {base_path}/swagger.json for swagger2) 

91 """ 

92 if not self.options.openapi_spec_path.endswith("json"): 

93 return 

94 

95 openapi_spec_path_yaml = self.options.openapi_spec_path[: -len("json")] + "yaml" 

96 logger.debug("Adding spec yaml: %s/%s", self.base_path, openapi_spec_path_yaml) 

97 self.router.add_route( 

98 methods=["GET"], 

99 path=openapi_spec_path_yaml, 

100 endpoint=self._get_openapi_yaml, 

101 ) 

102 

103 async def _get_openapi_json(self, request): 

104 # Yaml parses datetime objects when loading the spec, so we need our custom jsonifier to dump it 

105 jsonifier = Jsonifier() 

106 

107 return StarletteResponse( 

108 content=jsonifier.dumps(self._spec_for_prefix(request)), 

109 status_code=200, 

110 media_type="application/json", 

111 ) 

112 

113 async def _get_openapi_yaml(self, request): 

114 return StarletteResponse( 

115 content=yamldumper(self._spec_for_prefix(request)), 

116 status_code=200, 

117 media_type="text/yaml", 

118 ) 

119 

120 def add_swagger_ui(self): 

121 """ 

122 Adds swagger ui to {base_path}/ui/ 

123 """ 

124 console_ui_path = self.options.openapi_console_ui_path.strip().rstrip("/") 

125 logger.debug("Adding swagger-ui: %s%s/", self.base_path, console_ui_path) 

126 

127 for path in ( 

128 console_ui_path + "/", 

129 console_ui_path + "/index.html", 

130 ): 

131 self.router.add_route( 

132 methods=["GET"], path=path, endpoint=self._get_swagger_ui_home 

133 ) 

134 

135 if self.options.openapi_console_ui_config is not None: 

136 self.router.add_route( 

137 methods=["GET"], 

138 path=console_ui_path + "/swagger-ui-config.json", 

139 endpoint=self._get_swagger_ui_config, 

140 ) 

141 

142 # we have to add an explicit redirect instead of relying on the 

143 # normalize_path_middleware because we also serve static files 

144 # from this dir (below) 

145 

146 async def redirect(request): 

147 url = request.scope.get("root_path", "").rstrip("/") 

148 url += console_ui_path 

149 url += "/" 

150 return RedirectResponse(url=url) 

151 

152 self.router.add_route(methods=["GET"], path=console_ui_path, endpoint=redirect) 

153 

154 # this route will match and get a permission error when trying to 

155 # serve index.html, so we add the redirect above. 

156 self.router.mount( 

157 path=console_ui_path, 

158 app=StaticFiles(directory=str(self.options.openapi_console_ui_from_dir)), 

159 name="swagger_ui_static", 

160 ) 

161 

162 async def _get_swagger_ui_home(self, req): 

163 base_path = self._base_path_for_prefix(req) 

164 template_variables = { 

165 "request": req, 

166 "openapi_spec_url": (base_path + self.options.openapi_spec_path), 

167 **self.options.openapi_console_ui_index_template_variables, 

168 } 

169 if self.options.openapi_console_ui_config is not None: 

170 template_variables["configUrl"] = "swagger-ui-config.json" 

171 

172 return self._templates.TemplateResponse("index.j2", template_variables) 

173 

174 async def _get_swagger_ui_config(self, request): 

175 return StarletteResponse( 

176 status_code=200, 

177 media_type="application/json", 

178 content=json.dumps(self.options.openapi_console_ui_config), 

179 ) 

180 

181 

182class SwaggerUIMiddleware(SpecMiddleware): 

183 def __init__(self, app: ASGIApp, **kwargs) -> None: 

184 """Middleware that hosts a swagger UI. 

185 

186 :param app: app to wrap in middleware. 

187 """ 

188 self.app = app 

189 # Set default to pass unknown routes to next app 

190 self.router = Router(default=self.default_fn) 

191 

192 def add_api( 

193 self, 

194 specification: t.Union[pathlib.Path, str, dict], 

195 base_path: t.Optional[str] = None, 

196 arguments: t.Optional[dict] = None, 

197 **kwargs 

198 ) -> None: 

199 """Add an API to the router based on a OpenAPI spec. 

200 

201 :param specification: OpenAPI spec as dict or path to file. 

202 :param base_path: Base path where to add this API. 

203 :param arguments: Jinja arguments to replace in the spec. 

204 """ 

205 api = SwaggerUIAPI( 

206 specification, 

207 base_path=base_path, 

208 arguments=arguments, 

209 default=self.default_fn, 

210 **kwargs 

211 ) 

212 self.router.mount(api.base_path, app=api.router) 

213 

214 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

215 if scope["type"] != "http": 

216 await self.app(scope, receive, send) 

217 return 

218 

219 _original_scope.set(scope.copy()) # type: ignore 

220 await self.router(scope, receive, send) 

221 

222 async def default_fn(self, _scope: Scope, receive: Receive, send: Send) -> None: 

223 """ 

224 Callback to call next app as default when no matching route is found. 

225 

226 Unfortunately we cannot just pass the next app as default, since the router manipulates 

227 the scope when descending into mounts, losing information about the base path. Therefore, 

228 we use the original scope instead. 

229 

230 This is caused by https://github.com/encode/starlette/issues/1336. 

231 """ 

232 original_scope = _original_scope.get() 

233 await self.app(original_scope, receive, send)