Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/django/contrib/postgres/forms/array.py: 18%

1import copy 

2from itertools import chain 

3 

4from django import forms 

5from django.contrib.postgres.validators import ( 

6 ArrayMaxLengthValidator, 

7 ArrayMinLengthValidator, 

8) 

9from django.core.exceptions import ValidationError 

10from django.utils.translation import gettext_lazy as _ 

11 

12from ..utils import prefix_validation_error 

13 

14 

15class SimpleArrayField(forms.CharField): 

16 default_error_messages = { 

17 "item_invalid": _("Item %(nth)s in the array did not validate:"), 

18 } 

19 

20 def __init__( 

21 self, base_field, *, delimiter=",", max_length=None, min_length=None, **kwargs 

22 ): 

23 self.base_field = base_field 

24 self.delimiter = delimiter 

25 super().__init__(**kwargs) 

26 if min_length is not None: 

27 self.min_length = min_length 

28 self.validators.append(ArrayMinLengthValidator(int(min_length))) 

29 if max_length is not None: 

30 self.max_length = max_length 

31 self.validators.append(ArrayMaxLengthValidator(int(max_length))) 

32 

33 def clean(self, value): 

34 value = super().clean(value) 

35 return [self.base_field.clean(val) for val in value] 

36 

37 def prepare_value(self, value): 

38 if isinstance(value, list): 

39 return self.delimiter.join( 

40 str(self.base_field.prepare_value(v)) for v in value 

41 ) 

42 return value 

43 

44 def to_python(self, value): 

45 if isinstance(value, list): 

46 items = value 

47 elif value: 

48 items = value.split(self.delimiter) 

49 else: 

50 items = [] 

51 errors = [] 

52 values = [] 

53 for index, item in enumerate(items): 

54 try: 

55 values.append(self.base_field.to_python(item)) 

56 except ValidationError as error: 

57 errors.append( 

58 prefix_validation_error( 

59 error, 

60 prefix=self.error_messages["item_invalid"], 

61 code="item_invalid", 

62 params={"nth": index + 1}, 

63 ) 

64 ) 

65 if errors: 

66 raise ValidationError(errors) 

67 return values 

68 

69 def validate(self, value): 

70 super().validate(value) 

71 errors = [] 

72 for index, item in enumerate(value): 

73 try: 

74 self.base_field.validate(item) 

75 except ValidationError as error: 

76 errors.append( 

77 prefix_validation_error( 

78 error, 

79 prefix=self.error_messages["item_invalid"], 

80 code="item_invalid", 

81 params={"nth": index + 1}, 

82 ) 

83 ) 

84 if errors: 

85 raise ValidationError(errors) 

86 

87 def run_validators(self, value): 

88 super().run_validators(value) 

89 errors = [] 

90 for index, item in enumerate(value): 

91 try: 

92 self.base_field.run_validators(item) 

93 except ValidationError as error: 

94 errors.append( 

95 prefix_validation_error( 

96 error, 

97 prefix=self.error_messages["item_invalid"], 

98 code="item_invalid", 

99 params={"nth": index + 1}, 

100 ) 

101 ) 

102 if errors: 

103 raise ValidationError(errors) 

104 

105 def has_changed(self, initial, data): 

106 try: 

107 value = self.to_python(data) 

108 except ValidationError: 

109 pass 

110 else: 

111 if initial in self.empty_values and value in self.empty_values: 

112 return False 

113 return super().has_changed(initial, data) 

114 

115 

116class SplitArrayWidget(forms.Widget): 

117 template_name = "postgres/widgets/split_array.html" 

118 

119 def __init__(self, widget, size, **kwargs): 

120 self.widget = widget() if isinstance(widget, type) else widget 

121 self.size = size 

122 super().__init__(**kwargs) 

123 

124 @property 

125 def is_hidden(self): 

126 return self.widget.is_hidden 

127 

128 def value_from_datadict(self, data, files, name): 

129 return [ 

130 self.widget.value_from_datadict(data, files, "%s_%s" % (name, index)) 

131 for index in range(self.size) 

132 ] 

133 

134 def value_omitted_from_data(self, data, files, name): 

135 return all( 

136 self.widget.value_omitted_from_data(data, files, "%s_%s" % (name, index)) 

137 for index in range(self.size) 

138 ) 

139 

140 def id_for_label(self, id_): 

141 # See the comment for RadioSelect.id_for_label() 

142 if id_: 

143 id_ += "_0" 

144 return id_ 

145 

146 def get_context(self, name, value, attrs=None): 

147 attrs = {} if attrs is None else attrs 

148 context = super().get_context(name, value, attrs) 

149 if self.is_localized: 

150 self.widget.is_localized = self.is_localized 

151 value = value or [] 

152 context["widget"]["subwidgets"] = [] 

153 final_attrs = self.build_attrs(attrs) 

154 id_ = final_attrs.get("id") 

155 for i in range(max(len(value), self.size)): 

156 try: 

157 widget_value = value[i] 

158 except IndexError: 

159 widget_value = None 

160 if id_: 

161 final_attrs = {**final_attrs, "id": "%s_%s" % (id_, i)} 

162 context["widget"]["subwidgets"].append( 

163 self.widget.get_context(name + "_%s" % i, widget_value, final_attrs)[ 

164 "widget" 

165 ] 

166 ) 

167 return context 

168 

169 @property 

170 def media(self): 

171 return self.widget.media 

172 

173 def __deepcopy__(self, memo): 

174 obj = super().__deepcopy__(memo) 

175 obj.widget = copy.deepcopy(self.widget) 

176 return obj 

177 

178 @property 

179 def needs_multipart_form(self): 

180 return self.widget.needs_multipart_form 

181 

182 

183class SplitArrayField(forms.Field): 

184 default_error_messages = { 

185 "item_invalid": _("Item %(nth)s in the array did not validate:"), 

186 } 

187 

188 def __init__(self, base_field, size, *, remove_trailing_nulls=False, **kwargs): 

189 self.base_field = base_field 

190 self.size = size 

191 self.remove_trailing_nulls = remove_trailing_nulls 

192 widget = SplitArrayWidget(widget=base_field.widget, size=size) 

193 kwargs.setdefault("widget", widget) 

194 super().__init__(**kwargs) 

195 

196 def _remove_trailing_nulls(self, values): 

197 index = None 

198 if self.remove_trailing_nulls: 

199 for i, value in reversed(list(enumerate(values))): 

200 if value in self.base_field.empty_values: 

201 index = i 

202 else: 

203 break 

204 if index is not None: 

205 values = values[:index] 

206 return values, index 

207 

208 def to_python(self, value): 

209 value = super().to_python(value) 

210 return [self.base_field.to_python(item) for item in value] 

211 

212 def clean(self, value): 

213 cleaned_data = [] 

214 errors = [] 

215 if not any(value) and self.required: 

216 raise ValidationError(self.error_messages["required"]) 

217 max_size = max(self.size, len(value)) 

218 for index in range(max_size): 

219 item = value[index] 

220 try: 

221 cleaned_data.append(self.base_field.clean(item)) 

222 except ValidationError as error: 

223 errors.append( 

224 prefix_validation_error( 

225 error, 

226 self.error_messages["item_invalid"], 

227 code="item_invalid", 

228 params={"nth": index + 1}, 

229 ) 

230 ) 

231 cleaned_data.append(None) 

232 else: 

233 errors.append(None) 

234 cleaned_data, null_index = self._remove_trailing_nulls(cleaned_data) 

235 if null_index is not None: 

236 errors = errors[:null_index] 

237 errors = list(filter(None, errors)) 

238 if errors: 

239 raise ValidationError(list(chain.from_iterable(errors))) 

240 return cleaned_data 

241 

242 def has_changed(self, initial, data): 

243 try: 

244 data = self.to_python(data) 

245 except ValidationError: 

246 pass 

247 else: 

248 data, _ = self._remove_trailing_nulls(data) 

249 if initial in self.empty_values and data in self.empty_values: 

250 return False 

251 return super().has_changed(initial, data)