Coverage for src/backend/InvenTree/importer/serializers.py: 58%

1"""API serializers for the importer app.""" 

2 

3import json 

4 

5from django.core.exceptions import ValidationError 

6from django.utils.translation import gettext_lazy as _ 

7 

8from rest_framework import serializers 

9 

10import importer.models 

11import importer.registry 

12from InvenTree.serializers import ( 

13 InvenTreeAttachmentSerializerField, 

14 InvenTreeModelSerializer, 

15) 

16from users.serializers import UserSerializer 

17 

18 

19class DataImportColumnMapSerializer(InvenTreeModelSerializer): 

20 """Serializer for the DataImportColumnMap model.""" 

21 

22 class Meta: 

23 """Meta class options for the serializer.""" 

24 

25 model = importer.models.DataImportColumnMap 

26 fields = [ 

27 'pk', 

28 'session', 

29 'column', 

30 'field', 

31 'label', 

32 'description', 

33 'lookup_field', 

34 ] 

35 read_only_fields = ['field', 'session'] 

36 

37 label = serializers.CharField(read_only=True) 

38 description = serializers.CharField(read_only=True) 

39 

40 

41class DataImportSessionSerializer(InvenTreeModelSerializer): 

42 """Serializer for the DataImportSession model.""" 

43 

44 class Meta: 

45 """Meta class options for the serializer.""" 

46 

47 model = importer.models.DataImportSession 

48 fields = [ 

49 'pk', 

50 'timestamp', 

51 'data_file', 

52 'update_records', 

53 'model_type', 

54 'available_fields', 

55 'status', 

56 'user', 

57 'user_detail', 

58 'columns', 

59 'column_mappings', 

60 'field_defaults', 

61 'field_overrides', 

62 'field_filters', 

63 'row_count', 

64 'completed_row_count', 

65 'completed_row_count_history', 

66 'row_count_history', 

67 ] 

68 read_only_fields = ['pk', 'user', 'status', 'columns'] 

69 

70 def __init__(self, *args, **kwargs): 

71 """Override the constructor for the DataImportSession serializer.""" 

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

73 

74 self.fields['model_type'].choices = importer.registry.supported_model_options() 

75 

76 data_file = InvenTreeAttachmentSerializerField() 

77 

78 model_type = serializers.ChoiceField( 

79 required=True, 

80 allow_blank=False, 

81 choices=importer.registry.supported_model_options(), 

82 ) 

83 

84 available_fields = serializers.JSONField(read_only=True) 

85 

86 row_count = serializers.IntegerField(read_only=True) 

87 completed_row_count = serializers.IntegerField(read_only=True) 

88 

89 column_mappings = DataImportColumnMapSerializer(many=True, read_only=True) 

90 

91 user_detail = UserSerializer(source='user', read_only=True, many=False) 

92 

93 def validate_field_defaults(self, defaults): 

94 """De-stringify the field defaults.""" 

95 if defaults is None: 95 ↛ 96line 95 didn't jump to line 96 because the condition on line 95 was never true

96 return None 

97 

98 if type(defaults) is not dict: 98 ↛ 104line 98 didn't jump to line 104 because the condition on line 98 was always true

99 try: 

100 defaults = json.loads(str(defaults)) 

101 except: 

102 raise ValidationError(_('Invalid field defaults')) 

103 

104 return defaults 

105 

106 def validate_field_overrides(self, overrides): 

107 """De-stringify the field overrides.""" 

108 if overrides is None: 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true

109 return None 

110 

111 if type(overrides) is not dict: 111 ↛ 117line 111 didn't jump to line 117 because the condition on line 111 was always true

112 try: 

113 overrides = json.loads(str(overrides)) 

114 except: 

115 raise ValidationError(_('Invalid field overrides')) 

116 

117 return overrides 

118 

119 def validate_field_filters(self, filters): 

120 """De-stringify the field filters.""" 

121 if filters is None: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true

122 return None 

123 

124 if type(filters) is not dict: 124 ↛ 130line 124 didn't jump to line 130 because the condition on line 124 was always true

125 try: 

126 filters = json.loads(str(filters)) 

127 except: 

128 raise ValidationError(_('Invalid field filters')) 

129 

130 return filters 

131 

132 def create(self, validated_data): 

133 """Override create method for this serializer. 

134 

135 Attach user information based on provided session data. 

136 """ 

137 session = super().create(validated_data) 

138 

139 if request := self.context.get('request', None): 

140 session.user = request.user 

141 session.save() 

142 

143 return session 

144 

145 

146class DataImportRowSerializer(InvenTreeModelSerializer): 

147 """Serializer for the DataImportRow model.""" 

148 

149 class Meta: 

150 """Meta class options for the serializer.""" 

151 

152 model = importer.models.DataImportRow 

153 fields = [ 

154 'pk', 

155 'session', 

156 'row_index', 

157 'row_data', 

158 'data', 

159 'errors', 

160 'valid', 

161 'complete', 

162 ] 

163 

164 read_only_fields = [ 

165 'pk', 

166 'session', 

167 'row_index', 

168 'row_data', 

169 'errors', 

170 'valid', 

171 'complete', 

172 ] 

173 

174 

175class DataImportAcceptRowSerializer(serializers.Serializer): 

176 """Serializer for accepting rows of data.""" 

177 

178 class Meta: 

179 """Serializer meta options.""" 

180 

181 fields = ['rows'] 

182 

183 rows = serializers.PrimaryKeyRelatedField( 

184 queryset=importer.models.DataImportRow.objects.all(), 

185 many=True, 

186 required=True, 

187 label=_('Rows'), 

188 help_text=_('List of row IDs to accept'), 

189 ) 

190 

191 def validate_rows(self, rows): 

192 """Ensure that the provided rows are valid. 

193 

194 - Row must point to the same import session 

195 - Row must contain valid data 

196 - Row must not have already been completed 

197 """ 

198 session = self.context.get('session', None) 

199 

200 if not rows or len(rows) == 0: 200 ↛ 203line 200 didn't jump to line 203 because the condition on line 200 was always true

201 raise ValidationError(_('No rows provided')) 

202 

203 for row in rows: 

204 if row.session != session: 

205 raise ValidationError(_('Row does not belong to this session')) 

206 

207 if not row.valid: 

208 raise ValidationError(_('Row contains invalid data')) 

209 

210 if row.complete: 

211 raise ValidationError(_('Row has already been completed')) 

212 

213 return rows 

214 

215 def save(self): 

216 """Complete the provided rows.""" 

217 rows = self.validated_data['rows'] 

218 

219 request = self.context.get('request', None) 

220 

221 for row in rows: 

222 row.validate(commit=True, request=request) 

223 

224 if session := self.context.get('session', None): 

225 # ensure current state is available 

226 session.refresh_from_db() 

227 session.check_complete() 

228 

229 return rows