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.middleware.abstract import RoutedAPI, RoutedMiddleware 

18from connexion.middleware.lifespan import Lifespan 

19from connexion.operations import AbstractOperation 

20from connexion.resolver import Resolver 

21from connexion.uri_parsing import AbstractURIParser 

22 

23logger = logging.getLogger(__name__) 

24 

25 

26class AsyncOperation: 

27 def __init__( 

28 self, 

29 fn: t.Callable, 

30 jsonifier: Jsonifier, 

31 operation_id: str, 

32 pythonic_params: bool, 

33 ) -> None: 

34 self._fn = fn 

35 self.jsonifier = jsonifier 

36 self.operation_id = operation_id 

37 self.pythonic_params = pythonic_params 

38 functools.update_wrapper(self, fn) 

39 

40 @classmethod 

41 def from_operation( 

42 cls, 

43 operation: AbstractOperation, 

44 *, 

45 pythonic_params: bool, 

46 jsonifier: Jsonifier, 

47 ) -> "AsyncOperation": 

48 return cls( 

49 operation.function, 

50 jsonifier=jsonifier, 

51 operation_id=operation.operation_id, 

52 pythonic_params=pythonic_params, 

53 ) 

54 

55 @property 

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

57 decorator = StarletteDecorator( 

58 pythonic_params=self.pythonic_params, 

59 jsonifier=self.jsonifier, 

60 ) 

61 return decorator(self._fn) 

62 

63 async def __call__( 

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

65 ) -> StarletteResponse: 

66 response = await self.fn() 

67 return await response(scope, receive, send) 

68 

69 

70class AsyncApi(RoutedAPI[AsyncOperation]): 

71 def __init__( 

72 self, 

73 *args, 

74 pythonic_params: bool, 

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

76 **kwargs, 

77 ) -> None: 

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

79 self.pythonic_params = pythonic_params 

80 self.jsonifier = jsonifier or Jsonifier() 

81 self.router = Router() 

82 self.add_paths() 

83 

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

85 return AsyncOperation.from_operation( 

86 operation, pythonic_params=self.pythonic_params, jsonifier=self.jsonifier 

87 ) 

88 

89 

90class AsyncMiddlewareApp(RoutedMiddleware[AsyncApi]): 

91 

92 api_cls = AsyncApi 

93 

94 def __init__(self) -> None: 

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

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

97 self.router = Router() 

98 super().__init__(self.router) 

99 

100 def add_api(self, *args, **kwargs): 

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

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

103 return api 

104 

105 def add_url_rule( 

106 self, 

107 rule, 

108 endpoint: str = None, 

109 view_func: t.Callable = None, 

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

111 **options, 

112 ): 

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

114 

115 

116class AsyncApp(AbstractApp): 

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

118 based on starlette tools.""" 

119 

120 def __init__( 

121 self, 

122 import_name: str, 

123 *, 

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

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

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

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

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

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

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

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

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

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

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

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

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

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

138 ) -> None: 

139 """ 

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

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

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

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

144 :obj:`middleware.main.ConnexionmMiddleware.default_middlewares` 

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

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

147 the root path. 

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

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

150 Defaults to False. 

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

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

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

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

155 :class:`resolver.Resolver`. 

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

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

158 start. 

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

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

161 :param swagger_ui_options: A :class:`options.ConnexionOptions` instance with configuration 

162 options for the swagger ui. 

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

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

165 an impact on performance. Defaults to False. 

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

167 :obj:`validators.VALIDATOR_MAP`. 

168 """ 

169 self.middleware_app: AsyncMiddlewareApp = AsyncMiddlewareApp() 

170 

171 super().__init__( 

172 import_name, 

173 lifespan=lifespan, 

174 middlewares=middlewares, 

175 specification_dir=specification_dir, 

176 arguments=arguments, 

177 auth_all_paths=auth_all_paths, 

178 jsonifier=jsonifier, 

179 pythonic_params=pythonic_params, 

180 resolver=resolver, 

181 resolver_error=resolver_error, 

182 strict_validation=strict_validation, 

183 swagger_ui_options=swagger_ui_options, 

184 uri_parser_class=uri_parser_class, 

185 validate_responses=validate_responses, 

186 validator_map=validator_map, 

187 ) 

188 

189 def add_url_rule( 

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

191 ): 

192 self.middleware_app.add_url_rule( 

193 rule, endpoint=endpoint, view_func=view_func, **options 

194 ) 

195 

196 def add_error_handler( 

197 self, code_or_exception: t.Union[int, t.Type[Exception]], function: t.Callable 

198 ) -> None: 

199 self.middleware.add_error_handler(code_or_exception, function)