Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/django/utils/deprecation.py: 52%

1import asyncio 

2import inspect 

3import warnings 

4 

5from asgiref.sync import sync_to_async 

6 

7 

8class RemovedInDjango41Warning(DeprecationWarning): 

9 pass 

10 

11 

12class RemovedInDjango50Warning(PendingDeprecationWarning): 

13 pass 

14 

15 

16RemovedInNextVersionWarning = RemovedInDjango41Warning 

17 

18 

19class warn_about_renamed_method: 

20 def __init__( 

21 self, class_name, old_method_name, new_method_name, deprecation_warning 

22 ): 

23 self.class_name = class_name 

24 self.old_method_name = old_method_name 

25 self.new_method_name = new_method_name 

26 self.deprecation_warning = deprecation_warning 

27 

28 def __call__(self, f): 

29 def wrapped(*args, **kwargs): 

30 warnings.warn( 

31 "`%s.%s` is deprecated, use `%s` instead." 

32 % (self.class_name, self.old_method_name, self.new_method_name), 

33 self.deprecation_warning, 

34 2, 

35 ) 

36 return f(*args, **kwargs) 

37 

38 return wrapped 

39 

40 

41class RenameMethodsBase(type): 

42 """ 

43 Handles the deprecation paths when renaming a method. 

44 

45 It does the following: 

46 1) Define the new method if missing and complain about it. 

47 2) Define the old method if missing. 

48 3) Complain whenever an old method is called. 

49 

50 See #15363 for more details. 

51 """ 

52 

53 renamed_methods = () 

54 

55 def __new__(cls, name, bases, attrs): 

56 new_class = super().__new__(cls, name, bases, attrs) 

57 

58 for base in inspect.getmro(new_class): 

59 class_name = base.__name__ 

60 for renamed_method in cls.renamed_methods: 

61 old_method_name = renamed_method[0] 

62 old_method = base.__dict__.get(old_method_name) 

63 new_method_name = renamed_method[1] 

64 new_method = base.__dict__.get(new_method_name) 

65 deprecation_warning = renamed_method[2] 

66 wrapper = warn_about_renamed_method(class_name, *renamed_method) 

67 

68 # Define the new method if missing and complain about it 

69 if not new_method and old_method: 

70 warnings.warn( 

71 "`%s.%s` method should be renamed `%s`." 

72 % (class_name, old_method_name, new_method_name), 

73 deprecation_warning, 

74 2, 

75 ) 

76 setattr(base, new_method_name, old_method) 

77 setattr(base, old_method_name, wrapper(old_method)) 

78 

79 # Define the old method as a wrapped call to the new method. 

80 if not old_method and new_method: 

81 setattr(base, old_method_name, wrapper(new_method)) 

82 

83 return new_class 

84 

85 

86class DeprecationInstanceCheck(type): 

87 def __instancecheck__(self, instance): 

88 warnings.warn( 

89 "`%s` is deprecated, use `%s` instead." % (self.__name__, self.alternative), 

90 self.deprecation_warning, 

91 2, 

92 ) 

93 return super().__instancecheck__(instance) 

94 

95 

96class MiddlewareMixin: 

97 sync_capable = True 

98 async_capable = True 

99 

100 def __init__(self, get_response): 

101 if get_response is None: 101 ↛ 102line 101 didn't jump to line 102, because the condition on line 101 was never true

102 raise ValueError("get_response must be provided.") 

103 self.get_response = get_response 

104 self._async_check() 

105 super().__init__() 

106 

107 def __repr__(self): 

108 return "<%s get_response=%s>" % ( 

109 self.__class__.__qualname__, 

110 getattr( 

111 self.get_response, 

112 "__qualname__", 

113 self.get_response.__class__.__name__, 

114 ), 

115 ) 

116 

117 def _async_check(self): 

118 """ 

119 If get_response is a coroutine function, turns us into async mode so 

120 a thread is not consumed during a whole request. 

121 """ 

122 if asyncio.iscoroutinefunction(self.get_response): 122 ↛ 125line 122 didn't jump to line 125, because the condition on line 122 was never true

123 # Mark the class as async-capable, but do the actual switch 

124 # inside __call__ to avoid swapping out dunder methods 

125 self._is_coroutine = asyncio.coroutines._is_coroutine 

126 

127 def __call__(self, request): 

128 # Exit out to async mode, if needed 

129 if asyncio.iscoroutinefunction(self.get_response): 129 ↛ 130line 129 didn't jump to line 130, because the condition on line 129 was never true

130 return self.__acall__(request) 

131 response = None 

132 if hasattr(self, "process_request"): 

133 response = self.process_request(request) 

134 response = response or self.get_response(request) 

135 if hasattr(self, "process_response"): 

136 response = self.process_response(request, response) 

137 return response 

138 

139 async def __acall__(self, request): 

140 """ 

141 Async version of __call__ that is swapped in when an async request 

142 is running. 

143 """ 

144 response = None 

145 if hasattr(self, "process_request"): 

146 response = await sync_to_async( 

147 self.process_request, 

148 thread_sensitive=True, 

149 )(request) 

150 response = response or await self.get_response(request) 

151 if hasattr(self, "process_response"): 

152 response = await sync_to_async( 

153 self.process_response, 

154 thread_sensitive=True, 

155 )(request, response) 

156 return response