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

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 SyncResponseDecorator, 

20) 

21from connexion.frameworks.abstract import Framework 

22from connexion.frameworks.starlette import Starlette as StarletteFramework 

23from connexion.uri_parsing import AbstractURIParser 

24from connexion.utils import not_installed_error 

25 

26try: 

27 from connexion.frameworks.flask import Flask as FlaskFramework 

28except ImportError as e: 

29 _flask_not_installed_error = not_installed_error( 

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

31 ) 

32 FlaskFramework = _flask_not_installed_error # type: ignore 

33 

34 

35class BaseDecorator: 

36 """Base class for connexion decorators.""" 

37 

38 framework: t.Type[Framework] 

39 

40 def __init__( 

41 self, 

42 *, 

43 pythonic_params: bool = False, 

44 uri_parser_class: AbstractURIParser = None, 

45 jsonifier=json, 

46 ) -> None: 

47 self.pythonic_params = pythonic_params 

48 self.uri_parser_class = uri_parser_class 

49 self.jsonifier = jsonifier 

50 

51 self.arguments, self.has_kwargs = None, None 

52 

53 @property 

54 @abc.abstractmethod 

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

56 raise NotImplementedError 

57 

58 @property 

59 @abc.abstractmethod 

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

61 raise NotImplementedError 

62 

63 @property 

64 @abc.abstractmethod 

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

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

67 raise NotImplementedError 

68 

69 @property 

70 def uri_parser(self): 

71 uri_parser_class = self.uri_parser_class or operation.uri_parser_class 

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

73 

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

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

76 function = self._sync_async_decorator(function) 

77 

78 parameter_decorator = self._parameter_decorator_cls( 

79 framework=self.framework, 

80 pythonic_params=self.pythonic_params, 

81 ) 

82 function = parameter_decorator(function) 

83 

84 response_decorator = self._response_decorator_cls( 

85 framework=self.framework, 

86 jsonifier=self.jsonifier, 

87 ) 

88 function = response_decorator(function) 

89 

90 return function 

91 

92 @abc.abstractmethod 

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

94 raise NotImplementedError 

95 

96 

97class FlaskDecorator(BaseDecorator): 

98 """Decorator for usage with Flask. The parameter decorator works with a Flask request, 

99 and provides Flask datastructures to the view function. The response decorator returns 

100 a Flask response""" 

101 

102 framework = FlaskFramework 

103 

104 @property 

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

106 return SyncParameterDecorator 

107 

108 @property 

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

110 return SyncResponseDecorator 

111 

112 @property 

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

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

115 @functools.wraps(function) 

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

117 if asyncio.iscoroutinefunction(function): 

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

119 else: 

120 return function(*args, **kwargs) 

121 

122 return wrapper 

123 

124 return decorator 

125 

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

127 @functools.wraps(function) 

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

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

130 decorated_function = self.decorate(function) 

131 return decorated_function(request) 

132 

133 return wrapper 

134 

135 

136class ASGIDecorator(BaseDecorator): 

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

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

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

140 

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

142 

143 framework = StarletteFramework 

144 

145 @property 

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

147 return AsyncParameterDecorator 

148 

149 @property 

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

151 class NoResponseDecorator(BaseResponseDecorator): 

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

153 return lambda request: function(request) 

154 

155 return NoResponseDecorator 

156 

157 @property 

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

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

160 @functools.wraps(function) 

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

162 if asyncio.iscoroutinefunction(function): 

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

164 else: 

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

166 

167 return wrapper 

168 

169 return decorator 

170 

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

172 @functools.wraps(function) 

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

174 request = self.framework.get_request( 

175 uri_parser=self.uri_parser, scope=scope, receive=receive 

176 ) 

177 decorated_function = self.decorate(function) 

178 return await decorated_function(request) 

179 

180 return wrapper 

181 

182 

183class StarletteDecorator(ASGIDecorator): 

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

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

186 

187 The response decorator returns Starlette responses.""" 

188 

189 @property 

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

191 return AsyncResponseDecorator