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

1import abc 

2import functools 

3import inspect 

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: t.Optional[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 

106 framework = FlaskFramework 

107 

108 @property 

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

110 return SyncParameterDecorator 

111 

112 @property 

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

114 return NoResponseDecorator 

115 

116 @property 

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

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

119 @functools.wraps(function) 

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

121 if inspect.iscoroutinefunction(function): 

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

123 else: 

124 return function(*args, **kwargs) 

125 

126 return wrapper 

127 

128 return decorator 

129 

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

131 @functools.wraps(function) 

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

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

134 decorated_function = self.decorate(function) 

135 return decorated_function(request) 

136 

137 return wrapper 

138 

139 

140class FlaskDecorator(WSGIDecorator): 

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

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

143 

144 The response decorator returns Flask responses.""" 

145 

146 @property 

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

148 return SyncResponseDecorator 

149 

150 

151class ASGIDecorator(BaseDecorator): 

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

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

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

155 

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

157 """ 

158 

159 framework = StarletteFramework 

160 

161 @property 

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

163 return AsyncParameterDecorator 

164 

165 @property 

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

167 return NoResponseDecorator 

168 

169 @property 

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

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

172 @functools.wraps(function) 

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

174 if inspect.iscoroutinefunction(function): 

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

176 else: 

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

178 

179 return wrapper 

180 

181 return decorator 

182 

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

184 @functools.wraps(function) 

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

186 request = self.framework.get_request( 

187 uri_parser=self.uri_parser, scope=scope, receive=receive 

188 ) 

189 decorated_function = self.decorate(function) 

190 return await decorated_function(request) 

191 

192 return wrapper 

193 

194 

195class StarletteDecorator(ASGIDecorator): 

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

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

198 

199 The response decorator returns Starlette responses.""" 

200 

201 @property 

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

203 return AsyncResponseDecorator