Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/django/contrib/messages/storage/base.py: 41%

1from django.conf import settings 

2from django.contrib.messages import constants, utils 

3 

4LEVEL_TAGS = utils.get_level_tags() 

5 

6 

7class Message: 

8 """ 

9 Represent an actual message that can be stored in any of the supported 

10 storage classes (typically session- or cookie-based) and rendered in a view 

11 or template. 

12 """ 

13 

14 def __init__(self, level, message, extra_tags=None): 

15 self.level = int(level) 

16 self.message = message 

17 self.extra_tags = extra_tags 

18 

19 def _prepare(self): 

20 """ 

21 Prepare the message for serialization by forcing the ``message`` 

22 and ``extra_tags`` to str in case they are lazy translations. 

23 """ 

24 self.message = str(self.message) 

25 self.extra_tags = str(self.extra_tags) if self.extra_tags is not None else None 

26 

27 def __eq__(self, other): 

28 if not isinstance(other, Message): 

29 return NotImplemented 

30 return self.level == other.level and self.message == other.message 

31 

32 def __str__(self): 

33 return str(self.message) 

34 

35 @property 

36 def tags(self): 

37 return " ".join(tag for tag in [self.extra_tags, self.level_tag] if tag) 

38 

39 @property 

40 def level_tag(self): 

41 return LEVEL_TAGS.get(self.level, "") 

42 

43 

44class BaseStorage: 

45 """ 

46 This is the base backend for temporary message storage. 

47 

48 This is not a complete class; to be a usable storage backend, it must be 

49 subclassed and the two methods ``_get`` and ``_store`` overridden. 

50 """ 

51 

52 def __init__(self, request, *args, **kwargs): 

53 self.request = request 

54 self._queued_messages = [] 

55 self.used = False 

56 self.added_new = False 

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

58 

59 def __len__(self): 

60 return len(self._loaded_messages) + len(self._queued_messages) 

61 

62 def __iter__(self): 

63 self.used = True 

64 if self._queued_messages: 

65 self._loaded_messages.extend(self._queued_messages) 

66 self._queued_messages = [] 

67 return iter(self._loaded_messages) 

68 

69 def __contains__(self, item): 

70 return item in self._loaded_messages or item in self._queued_messages 

71 

72 def __repr__(self): 

73 return f"<{self.__class__.__qualname__}: request={self.request!r}>" 

74 

75 @property 

76 def _loaded_messages(self): 

77 """ 

78 Return a list of loaded messages, retrieving them first if they have 

79 not been loaded yet. 

80 """ 

81 if not hasattr(self, "_loaded_data"): 

82 messages, all_retrieved = self._get() 

83 self._loaded_data = messages or [] 

84 return self._loaded_data 

85 

86 def _get(self, *args, **kwargs): 

87 """ 

88 Retrieve a list of stored messages. Return a tuple of the messages 

89 and a flag indicating whether or not all the messages originally 

90 intended to be stored in this storage were, in fact, stored and 

91 retrieved; e.g., ``(messages, all_retrieved)``. 

92 

93 **This method must be implemented by a subclass.** 

94 

95 If it is possible to tell if the backend was not used (as opposed to 

96 just containing no messages) then ``None`` should be returned in 

97 place of ``messages``. 

98 """ 

99 raise NotImplementedError( 

100 "subclasses of BaseStorage must provide a _get() method" 

101 ) 

102 

103 def _store(self, messages, response, *args, **kwargs): 

104 """ 

105 Store a list of messages and return a list of any messages which could 

106 not be stored. 

107 

108 One type of object must be able to be stored, ``Message``. 

109 

110 **This method must be implemented by a subclass.** 

111 """ 

112 raise NotImplementedError( 

113 "subclasses of BaseStorage must provide a _store() method" 

114 ) 

115 

116 def _prepare_messages(self, messages): 

117 """ 

118 Prepare a list of messages for storage. 

119 """ 

120 for message in messages: 120 ↛ 121line 120 didn't jump to line 121, because the loop on line 120 never started

121 message._prepare() 

122 

123 def update(self, response): 

124 """ 

125 Store all unread messages. 

126 

127 If the backend has yet to be iterated, store previously stored messages 

128 again. Otherwise, only store messages added after the last iteration. 

129 """ 

130 self._prepare_messages(self._queued_messages) 

131 if self.used: 131 ↛ 132line 131 didn't jump to line 132, because the condition on line 131 was never true

132 return self._store(self._queued_messages, response) 

133 elif self.added_new: 133 ↛ 134line 133 didn't jump to line 134, because the condition on line 133 was never true

134 messages = self._loaded_messages + self._queued_messages 

135 return self._store(messages, response) 

136 

137 def add(self, level, message, extra_tags=""): 

138 """ 

139 Queue a message to be stored. 

140 

141 The message is only queued if it contained something and its level is 

142 not less than the recording level (``self.level``). 

143 """ 

144 if not message: 

145 return 

146 # Check that the message level is not less than the recording level. 

147 level = int(level) 

148 if level < self.level: 

149 return 

150 # Add the message. 

151 self.added_new = True 

152 message = Message(level, message, extra_tags=extra_tags) 

153 self._queued_messages.append(message) 

154 

155 def _get_level(self): 

156 """ 

157 Return the minimum recorded level. 

158 

159 The default level is the ``MESSAGE_LEVEL`` setting. If this is 

160 not found, the ``INFO`` level is used. 

161 """ 

162 if not hasattr(self, "_level"): 

163 self._level = getattr(settings, "MESSAGE_LEVEL", constants.INFO) 

164 return self._level 

165 

166 def _set_level(self, value=None): 

167 """ 

168 Set a custom minimum recorded level. 

169 

170 If set to ``None``, the default level will be used (see the 

171 ``_get_level`` method). 

172 """ 

173 if value is None and hasattr(self, "_level"): 

174 del self._level 

175 else: 

176 self._level = int(value) 

177 

178 level = property(_get_level, _set_level, _set_level)