Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/connexion/decorators/main.py: 52%

1import abc 

2import asyncio 

3import functools 

4import json 

5import typing as t 

6 

7from asgiref.sync import async_to_sync 

8from starlette.concurrency import run_in_threadpool 

9 

10from connexion.context import operation, receive, scope 

11from connexion.decorators.parameter import ( 

12 AsyncParameterDecorator, 

13 BaseParameterDecorator, 

14 SyncParameterDecorator, 

15) 

16from connexion.decorators.response import ( 

17 AsyncResponseDecorator, 

18 BaseResponseDecorator, 

19 NoResponseDecorator, 

20 SyncResponseDecorator, 

21) 

22from connexion.frameworks.abstract import Framework 

23from connexion.frameworks.starlette import Starlette as StarletteFramework 

24from connexion.uri_parsing import AbstractURIParser 

25from connexion.utils import not_installed_error 

26 

27try: 

28 from connexion.frameworks.flask import Flask as FlaskFramework 

29except ImportError as e: 

30 _flask_not_installed_error = not_installed_error( 

31 e, msg="Please install connexion using the 'flask' extra" 

32 ) 

33 FlaskFramework = _flask_not_installed_error # type: ignore 

34 

35 

36class BaseDecorator: 

37 """Base class for connexion decorators.""" 

38 

39 framework: t.Type[Framework] 

40 

41 def __init__( 

42 self, 

43 *, 

44 pythonic_params: bool = False, 

45 uri_parser_class: AbstractURIParser = None, 

46 jsonifier=json, 

47 ) -> None: 

48 self.pythonic_params = pythonic_params 

49 self.uri_parser_class = uri_parser_class 

50 self.jsonifier = jsonifier 

51 

52 self.arguments, self.has_kwargs = None, None 

53 

54 @property 

55 @abc.abstractmethod 

56 def _parameter_decorator_cls(self) -> t.Type[BaseParameterDecorator]: 

57 raise NotImplementedError 

58 

59 @property 

60 @abc.abstractmethod 

61 def _response_decorator_cls(self) -> t.Type[BaseResponseDecorator]: 

62 raise NotImplementedError 

63 

64 @property 

65 @abc.abstractmethod 

66 def _sync_async_decorator(self) -> t.Callable[[t.Callable], t.Callable]: 

67 """Decorator to translate between sync and async functions.""" 

68 raise NotImplementedError 

69 

70 @property 

71 def uri_parser(self): 

72 uri_parser_class = self.uri_parser_class or operation.uri_parser_class 

73 return uri_parser_class(operation.parameters, operation.body_definition()) 

74 

75 def decorate(self, function: t.Callable) -> t.Callable: 

76 """Decorate a function with decorators based on the operation.""" 

77 function = self._sync_async_decorator(function) 

78 

79 parameter_decorator = self._parameter_decorator_cls( 

80 framework=self.framework, 

81 pythonic_params=self.pythonic_params, 

82 ) 

83 function = parameter_decorator(function) 

84 

85 response_decorator = self._response_decorator_cls( 

86 framework=self.framework, 

87 jsonifier=self.jsonifier, 

88 ) 

89 function = response_decorator(function) 

90 

91 return function 

92 

93 @abc.abstractmethod 

94 def __call__(self, function: t.Callable) -> t.Callable: 

95 raise NotImplementedError 

96 

97 

98class WSGIDecorator(BaseDecorator): 

99 """Decorator for usage with WSGI apps. The parameter decorator works with a Flask request, 

100 and provides Flask datastructures to the view function. This works for any WSGI app, since 

101 we get the request via the connexion context provided by WSGI middleware. 

102 

103 This decorator does not parse responses, but passes them directly to the WSGI App.""" 

104 

105 framework = FlaskFramework 

106 

107 @property 

108 def _parameter_decorator_cls(self) -> t.Type[SyncParameterDecorator]: 

109 return SyncParameterDecorator 

110 

111 @property 

112 def _response_decorator_cls(self) -> t.Type[BaseResponseDecorator]: 

113 return NoResponseDecorator 

114 

115 @property 

116 def _sync_async_decorator(self) -> t.Callable[[t.Callable], t.Callable]: 

117 def decorator(function: t.Callable) -> t.Callable: 

118 @functools.wraps(function) 

119 def wrapper(*args, **kwargs) -> t.Callable: 

120 if asyncio.iscoroutinefunction(function): 

121 return async_to_sync(function)(*args, **kwargs) 

122 else: 

123 return function(*args, **kwargs) 

124 

125 return wrapper 

126 

127 return decorator 

128 

129 def __call__(self, function: t.Callable) -> t.Callable: 

130 @functools.wraps(function) 

131 def wrapper(*args, **kwargs): 

132 request = self.framework.get_request(uri_parser=self.uri_parser) 

133 decorated_function = self.decorate(function) 

134 return decorated_function(request) 

135 

136 return wrapper 

137 

138 

139class FlaskDecorator(WSGIDecorator): 

140 """Decorator for usage with Connexion or Flask apps. The parameter decorator works with a 

141 Flask request, and provides Flask datastructures to the view function. 

142 

143 The response decorator returns Flask responses.""" 

144 

145 @property 

146 def _response_decorator_cls(self) -> t.Type[SyncResponseDecorator]: 

147 return SyncResponseDecorator 

148 

149 

150class ASGIDecorator(BaseDecorator): 

151 """Decorator for usage with ASGI apps. The parameter decorator works with a Starlette request, 

152 and provides Starlette datastructures to the view function. This works for any ASGI app, since 

153 we get the request via the connexion context provided by ASGI middleware. 

154 

155 This decorator does not parse responses, but passes them directly to the ASGI App.""" 

156 

157 framework = StarletteFramework 

158 

159 @property 

160 def _parameter_decorator_cls(self) -> t.Type[AsyncParameterDecorator]: 

161 return AsyncParameterDecorator 

162 

163 @property 

164 def _response_decorator_cls(self) -> t.Type[BaseResponseDecorator]: 

165 return NoResponseDecorator 

166 

167 @property 

168 def _sync_async_decorator(self) -> t.Callable[[t.Callable], t.Callable]: 

169 def decorator(function: t.Callable) -> t.Callable: 

170 @functools.wraps(function) 

171 async def wrapper(*args, **kwargs): 

172 if asyncio.iscoroutinefunction(function): 

173 return await function(*args, **kwargs) 

174 else: 

175 return await run_in_threadpool(function, *args, **kwargs) 

176 

177 return wrapper 

178 

179 return decorator 

180 

181 def __call__(self, function: t.Callable) -> t.Callable: 

182 @functools.wraps(function) 

183 async def wrapper(*args, **kwargs): 

184 request = self.framework.get_request( 

185 uri_parser=self.uri_parser, scope=scope, receive=receive 

186 ) 

187 decorated_function = self.decorate(function) 

188 return await decorated_function(request) 

189 

190 return wrapper 

191 

192 

193class StarletteDecorator(ASGIDecorator): 

194 """Decorator for usage with Connexion or Starlette apps. The parameter decorator works with a 

195 Starlette request, and provides Starlette datastructures to the view function. 

196 

197 The response decorator returns Starlette responses.""" 

198 

199 @property 

200 def _response_decorator_cls(self) -> t.Type[AsyncResponseDecorator]: 

201 return AsyncResponseDecorator