Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/connexion/apps/asynchronous.py: 56%

1""" 

2This module defines a native connexion asynchronous application. 

3""" 

4 

5import functools 

6import logging 

7import pathlib 

8import typing as t 

9 

10from starlette.responses import Response as StarletteResponse 

11from starlette.routing import Router 

12from starlette.types import Receive, Scope, Send 

13 

14from connexion.apps.abstract import AbstractApp 

15from connexion.decorators import StarletteDecorator 

16from connexion.jsonifier import Jsonifier 

17from connexion.lifecycle import ConnexionRequest, ConnexionResponse 

18from connexion.middleware.abstract import RoutedAPI, RoutedMiddleware 

19from connexion.middleware.lifespan import Lifespan 

20from connexion.operations import AbstractOperation 

21from connexion.options import SwaggerUIOptions 

22from connexion.resolver import Resolver 

23from connexion.types import MaybeAwaitable 

24from connexion.uri_parsing import AbstractURIParser 

25 

26logger = logging.getLogger(__name__) 

27 

28 

29class AsyncOperation: 

30 def __init__( 

31 self, 

32 fn: t.Callable, 

33 jsonifier: Jsonifier, 

34 operation_id: str, 

35 pythonic_params: bool, 

36 ) -> None: 

37 self._fn = fn 

38 self.jsonifier = jsonifier 

39 self.operation_id = operation_id 

40 self.pythonic_params = pythonic_params 

41 functools.update_wrapper(self, fn) 

42 

43 @classmethod 

44 def from_operation( 

45 cls, 

46 operation: AbstractOperation, 

47 *, 

48 pythonic_params: bool, 

49 jsonifier: Jsonifier, 

50 ) -> "AsyncOperation": 

51 return cls( 

52 operation.function, 

53 jsonifier=jsonifier, 

54 operation_id=operation.operation_id, 

55 pythonic_params=pythonic_params, 

56 ) 

57 

58 @property 

59 def fn(self) -> t.Callable: 

60 decorator = StarletteDecorator( 

61 pythonic_params=self.pythonic_params, 

62 jsonifier=self.jsonifier, 

63 ) 

64 return decorator(self._fn) 

65 

66 async def __call__( 

67 self, scope: Scope, receive: Receive, send: Send 

68 ) -> StarletteResponse: 

69 response = await self.fn() 

70 return await response(scope, receive, send) 

71 

72 

73class AsyncApi(RoutedAPI[AsyncOperation]): 

74 def __init__( 

75 self, 

76 *args, 

77 pythonic_params: bool, 

78 jsonifier: t.Optional[Jsonifier] = None, 

79 **kwargs, 

80 ) -> None: 

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

82 self.pythonic_params = pythonic_params 

83 self.jsonifier = jsonifier or Jsonifier() 

84 self.router = Router() 

85 self.add_paths() 

86 

87 def make_operation(self, operation: AbstractOperation) -> AsyncOperation: 

88 return AsyncOperation.from_operation( 

89 operation, pythonic_params=self.pythonic_params, jsonifier=self.jsonifier 

90 ) 

91 

92 

93class AsyncASGIApp(RoutedMiddleware[AsyncApi]): 

94 

95 api_cls = AsyncApi 

96 

97 def __init__(self) -> None: 

98 self.apis: t.Dict[str, t.List[AsyncApi]] = {} 

99 self.operations: t.Dict[str, AsyncOperation] = {} 

100 self.router = Router() 

101 super().__init__(self.router) 

102 

103 def add_api(self, *args, name: str = None, **kwargs): 

104 api = super().add_api(*args, **kwargs) 

105 

106 if name is not None: 

107 self.router.mount(api.base_path, api.router, name=name) 

108 else: 

109 self.router.mount(api.base_path, api.router) 

110 return api 

111 

112 def add_url_rule( 

113 self, 

114 rule, 

115 endpoint: str = None, 

116 view_func: t.Callable = None, 

117 methods: t.List[str] = None, 

118 **options, 

119 ): 

120 self.router.add_route(rule, endpoint=view_func, name=endpoint, methods=methods) 

121 

122 

123class AsyncApp(AbstractApp): 

124 """Connexion Application based on ConnexionMiddleware wrapping a async Connexion application 

125 based on starlette tools.""" 

126 

127 def __init__( 

128 self, 

129 import_name: str, 

130 *, 

131 lifespan: t.Optional[Lifespan] = None, 

132 middlewares: t.Optional[list] = None, 

133 specification_dir: t.Union[pathlib.Path, str] = "", 

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

135 auth_all_paths: t.Optional[bool] = None, 

136 jsonifier: t.Optional[Jsonifier] = None, 

137 pythonic_params: t.Optional[bool] = None, 

138 resolver: t.Optional[t.Union[Resolver, t.Callable]] = None, 

139 resolver_error: t.Optional[int] = None, 

140 strict_validation: t.Optional[bool] = None, 

141 swagger_ui_options: t.Optional[SwaggerUIOptions] = None, 

142 uri_parser_class: t.Optional[AbstractURIParser] = None, 

143 validate_responses: t.Optional[bool] = None, 

144 validator_map: t.Optional[dict] = None, 

145 security_map: t.Optional[dict] = None, 

146 ) -> None: 

147 """ 

148 :param import_name: The name of the package or module that this object belongs to. If you 

149 are using a single module, __name__ is always the correct value. If you however are 

150 using a package, it’s usually recommended to hardcode the name of your package there. 

151 :param lifespan: A lifespan context function, which can be used to perform startup and 

152 shutdown tasks. 

153 :param middlewares: The list of middlewares to wrap around the application. Defaults to 

154 :obj:`middleware.main.ConnexionMiddleware.default_middlewares` 

155 :param specification_dir: The directory holding the specification(s). The provided path 

156 should either be absolute or relative to the root path of the application. Defaults to 

157 the root path. 

158 :param arguments: Arguments to substitute the specification using Jinja. 

159 :param auth_all_paths: whether to authenticate not paths not defined in the specification. 

160 Defaults to False. 

161 :param jsonifier: Custom jsonifier to overwrite json encoding for json responses. 

162 :param pythonic_params: When True, CamelCase parameters are converted to snake_case and an 

163 underscore is appended to any shadowed built-ins. Defaults to False. 

164 :param resolver: Callable that maps operationId to a function or instance of 

165 :class:`resolver.Resolver`. 

166 :param resolver_error: Error code to return for operations for which the operationId could 

167 not be resolved. If no error code is provided, the application will fail when trying to 

168 start. 

169 :param strict_validation: When True, extra form or query parameters not defined in the 

170 specification result in a validation error. Defaults to False. 

171 :param swagger_ui_options: Instance of :class:`options.ConnexionOptions` with 

172 configuration options for the swagger ui. 

173 :param uri_parser_class: Class to use for uri parsing. See :mod:`uri_parsing`. 

174 :param validate_responses: Whether to validate responses against the specification. This has 

175 an impact on performance. Defaults to False. 

176 :param validator_map: A dictionary of validators to use. Defaults to 

177 :obj:`validators.VALIDATOR_MAP`. 

178 :param security_map: A dictionary of security handlers to use. Defaults to 

179 :obj:`security.SECURITY_HANDLERS` 

180 """ 

181 self._middleware_app: AsyncASGIApp = AsyncASGIApp() 

182 

183 super().__init__( 

184 import_name, 

185 lifespan=lifespan, 

186 middlewares=middlewares, 

187 specification_dir=specification_dir, 

188 arguments=arguments, 

189 auth_all_paths=auth_all_paths, 

190 jsonifier=jsonifier, 

191 pythonic_params=pythonic_params, 

192 resolver=resolver, 

193 resolver_error=resolver_error, 

194 strict_validation=strict_validation, 

195 swagger_ui_options=swagger_ui_options, 

196 uri_parser_class=uri_parser_class, 

197 validate_responses=validate_responses, 

198 validator_map=validator_map, 

199 security_map=security_map, 

200 ) 

201 

202 def add_url_rule( 

203 self, rule, endpoint: str = None, view_func: t.Callable = None, **options 

204 ): 

205 self._middleware_app.add_url_rule( 

206 rule, endpoint=endpoint, view_func=view_func, **options 

207 ) 

208 

209 def add_error_handler( 

210 self, 

211 code_or_exception: t.Union[int, t.Type[Exception]], 

212 function: t.Callable[ 

213 [ConnexionRequest, Exception], MaybeAwaitable[ConnexionResponse] 

214 ], 

215 ) -> None: 

216 self.middleware.add_error_handler(code_or_exception, function)