Coverage for extras/api/customfields.py: 46%

1from django.core.exceptions import ValidationError as DjangoValidationError 

2from django.utils.translation import gettext as _ 

3from drf_spectacular.types import OpenApiTypes 

4from drf_spectacular.utils import extend_schema_field 

5from rest_framework.fields import Field 

6from rest_framework.serializers import ListSerializer, ValidationError 

7 

8from extras.choices import CustomFieldTypeChoices 

9from extras.constants import CUSTOMFIELD_EMPTY_VALUES 

10from extras.models import CustomField 

11from utilities.api import get_serializer_for_model 

12from utilities.forms.fields import LaxURLField 

13 

14# 

15# Custom fields 

16# 

17 

18 

19class CustomFieldDefaultValues: 

20 """ 

21 Return a dictionary of all CustomFields assigned to the parent model and their default values. 

22 """ 

23 requires_context = True 

24 

25 def __call__(self, serializer_field): 

26 self.model = serializer_field.parent.Meta.model 

27 

28 # Populate the default value for each CustomField on the model 

29 value = {} 

30 for field in CustomField.objects.get_for_model(self.model): 30 ↛ 31line 30 didn't jump to line 31 because the loop on line 30 never started

31 if field.default is not None: 

32 value[field.name] = field.default 

33 else: 

34 value[field.name] = None 

35 

36 return value 

37 

38 

39@extend_schema_field(OpenApiTypes.OBJECT) 

40class CustomFieldsDataField(Field): 

41 

42 def _get_custom_fields(self): 

43 """ 

44 Cache CustomFields assigned to this model to avoid redundant database queries 

45 """ 

46 if not hasattr(self, '_custom_fields'): 

47 self._custom_fields = CustomField.objects.get_for_model(self.parent.Meta.model) 

48 return self._custom_fields 

49 

50 def to_representation(self, obj): 

51 # TODO: Fix circular import 

52 from utilities.api import get_serializer_for_model 

53 data = {} 

54 cache = self.parent.context.get('cf_object_cache') 

55 

56 for cf in self._get_custom_fields(): 56 ↛ 57line 56 didn't jump to line 57 because the loop on line 56 never started

57 if cache is not None and cf.type in ( 

58 CustomFieldTypeChoices.TYPE_OBJECT, 

59 CustomFieldTypeChoices.TYPE_MULTIOBJECT, 

60 ): 

61 raw = obj.get(cf.name) 

62 if raw is None: 

63 value = None 

64 elif cf.type == CustomFieldTypeChoices.TYPE_OBJECT: 

65 model = cf.related_object_type.model_class() 

66 value = cache.get((model, raw)) 

67 else: 

68 model = cf.related_object_type.model_class() 

69 value = [cache[(model, pk)] for pk in raw if (model, pk) in cache] or None 

70 else: 

71 value = cf.deserialize(obj.get(cf.name)) 

72 

73 if value is not None and cf.type == CustomFieldTypeChoices.TYPE_OBJECT: 

74 serializer = get_serializer_for_model(cf.related_object_type.model_class()) 

75 value = serializer(value, nested=True, context=self.parent.context).data 

76 elif value is not None and cf.type == CustomFieldTypeChoices.TYPE_MULTIOBJECT: 

77 serializer = get_serializer_for_model(cf.related_object_type.model_class()) 

78 value = serializer(value, nested=True, many=True, context=self.parent.context).data 

79 elif cf.type in (CustomFieldTypeChoices.TYPE_SELECT, CustomFieldTypeChoices.TYPE_MULTISELECT): 

80 value = cf.resolve_selection_value(value) 

81 data[cf.name] = value 

82 

83 return data 

84 

85 def to_internal_value(self, data): 

86 if type(data) is not dict: 

87 raise ValidationError( 

88 "Invalid data format. Custom field data must be passed as a dictionary mapping field names to their " 

89 "values." 

90 ) 

91 

92 custom_fields = {cf.name: cf for cf in self._get_custom_fields()} 

93 

94 # Reject any unknown custom field names 

95 invalid_fields = set(data) - set(custom_fields) 

96 if invalid_fields: 

97 raise ValidationError({ 

98 field: _("Custom field '{name}' does not exist for this object type.").format(name=field) 

99 for field in sorted(invalid_fields) 

100 }) 

101 

102 # Serialize object and multi-object values 

103 for cf in custom_fields.values(): 103 ↛ 104line 103 didn't jump to line 104 because the loop on line 103 never started

104 if cf.name in data and data[cf.name] not in CUSTOMFIELD_EMPTY_VALUES and cf.type in ( 

105 CustomFieldTypeChoices.TYPE_OBJECT, 

106 CustomFieldTypeChoices.TYPE_MULTIOBJECT 

107 ): 

108 serializer_class = get_serializer_for_model(cf.related_object_type.model_class()) 

109 many = cf.type == CustomFieldTypeChoices.TYPE_MULTIOBJECT 

110 serializer = serializer_class(data=data[cf.name], nested=True, many=many, context=self.parent.context) 

111 if serializer.is_valid(): 

112 data[cf.name] = [obj['id'] for obj in serializer.data] if many else serializer.data['id'] 

113 else: 

114 raise ValidationError(_("Unknown related object(s): {name}").format(name=data[cf.name])) 

115 

116 # Normalize URL values the same way the UI does (LaxURLField with assume_scheme='https'), so a 

117 # schemeless value (e.g. "example.com") is stored as an absolute URL ("https://example.com"). 

118 # Malformed values are left untouched for CustomField.validate() to report. 

119 elif cf.type == CustomFieldTypeChoices.TYPE_URL and isinstance(data.get(cf.name), str) and data[cf.name]: 

120 try: 

121 data[cf.name] = LaxURLField(assume_scheme='https').to_python(data[cf.name]) 

122 except DjangoValidationError: 

123 pass 

124 

125 # If updating an existing instance, start with existing custom_field_data 

126 if self.parent.instance: 

127 data = {**self.parent.instance.custom_field_data, **data} 

128 

129 return data 

130 

131 

132class CustomFieldListSerializer(ListSerializer): 

133 """ 

134 ListSerializer that pre-fetches all OBJECT/MULTIOBJECT custom field related objects 

135 in bulk before per-item serialization. 

136 """ 

137 def to_representation(self, data): 

138 cf_field = self.child.fields.get('custom_fields') 

139 if isinstance(cf_field, CustomFieldsDataField): 

140 object_type_cfs = [ 

141 cf for cf in cf_field._get_custom_fields() 

142 if cf.type in (CustomFieldTypeChoices.TYPE_OBJECT, CustomFieldTypeChoices.TYPE_MULTIOBJECT) 

143 ] 

144 cache = {} 

145 for cf in object_type_cfs: 145 ↛ 146line 145 didn't jump to line 146 because the loop on line 145 never started

146 model = cf.related_object_type.model_class() 

147 pks = set() 

148 for item in data: 

149 raw = item.custom_field_data.get(cf.name) 

150 if raw is not None: 

151 if cf.type == CustomFieldTypeChoices.TYPE_MULTIOBJECT: 

152 pks.update(raw) 

153 else: 

154 pks.add(raw) 

155 for obj in model.objects.filter(pk__in=pks): 

156 cache[(model, obj.pk)] = obj 

157 self.child.context['cf_object_cache'] = cache 

158 return super().to_representation(data)