Coverage for netbox/api/fields.py: 85%
104 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1from django.core.exceptions import ObjectDoesNotExist
2from django.db.backends.postgresql.psycopg_any import NumericRange
3from django.utils.translation import gettext as _
4from drf_spectacular.types import OpenApiTypes
5from drf_spectacular.utils import extend_schema_field
6from netaddr import IPNetwork
7from rest_framework import serializers
8from rest_framework.exceptions import ValidationError
9from rest_framework.relations import PrimaryKeyRelatedField, RelatedField
11from utilities.data import get_inclusive_integer_range_bounds
13__all__ = (
14 'AttributesField',
15 'ChoiceField',
16 'ContentTypeField',
17 'IPNetworkSerializer',
18 'IntegerRangeSerializer',
19 'RelatedObjectCountField',
20 'SerializedPKRelatedField',
21)
24class ChoiceField(serializers.Field):
25 """
26 Represent a ChoiceField as {'value': <DB value>, 'label': <string>}. Accepts a single value on write.
28 :param choices: An iterable of choices in the form (value, key).
29 :param allow_blank: Allow blank values in addition to the listed choices.
30 """
31 def __init__(self, choices, allow_blank=False, **kwargs):
32 self.choiceset = choices
33 self.allow_blank = allow_blank
34 self._choices = dict()
36 # Unpack grouped choices
37 for k, v in choices:
38 if type(v) in [list, tuple]:
39 for k2, v2 in v:
40 self._choices[k2] = v2
41 else:
42 self._choices[k] = v
44 super().__init__(**kwargs)
46 def validate_empty_values(self, data):
47 # Convert null to an empty string unless allow_null == True
48 if data is None:
49 if self.allow_null:
50 return True, None
51 data = ''
52 return super().validate_empty_values(data)
54 def to_representation(self, obj):
55 if obj != '':
56 # Use an empty string in place of the choice label if it cannot be resolved (i.e. because a previously
57 # configured choice has been removed from FIELD_CHOICES).
58 return {
59 'value': obj,
60 'label': self._choices.get(obj, ''),
61 }
62 return None
64 def to_internal_value(self, data):
65 if data == '':
66 if self.allow_blank: 66 ↛ 68line 66 didn't jump to line 68 because the condition on line 66 was always true
67 return data
68 raise ValidationError(_("This field may not be blank."))
70 # Provide an explicit error message if the request is trying to write a dict or list
71 if isinstance(data, (dict, list)):
72 raise ValidationError(
73 _('Value must be passed directly (e.g. "foo": 123); do not use a dictionary or list.')
74 )
76 # Check for string representations of boolean/integer values
77 if hasattr(data, 'lower'):
78 if data.lower() == 'true':
79 data = True
80 elif data.lower() == 'false': 80 ↛ 81line 80 didn't jump to line 81 because the condition on line 80 was never true
81 data = False
82 else:
83 try:
84 data = int(data)
85 except ValueError:
86 pass
88 try:
89 if data in self._choices:
90 return data
91 except TypeError: # Input is an unhashable type
92 pass
94 raise ValidationError(_("{value} is not a valid choice.").format(value=data))
96 @property
97 def choices(self):
98 return self._choices
101@extend_schema_field(OpenApiTypes.STR)
102class ContentTypeField(RelatedField):
103 """
104 Represent a ContentType as '<app_label>.<model>'
105 """
106 default_error_messages = {
107 "does_not_exist": _("Invalid content type: {content_type}"),
108 "invalid": _("Invalid value. Specify a content type as '<app_label>.<model_name>'."),
109 }
111 def to_internal_value(self, data):
112 try:
113 app_label, model = data.split('.')
114 # Scoped to the field's declared queryset, not the raw ContentType table (#22748).
115 return self.get_queryset().get(app_label=app_label, model=model)
116 except ObjectDoesNotExist:
117 self.fail('does_not_exist', content_type=data)
118 except (AttributeError, TypeError, ValueError):
119 self.fail('invalid')
121 def to_representation(self, obj):
122 return f"{obj.app_label}.{obj.model}"
125class IPNetworkSerializer(serializers.Serializer):
126 """
127 Representation of an IP network value (e.g. 192.0.2.0/24).
128 """
129 def to_representation(self, instance):
130 return str(instance)
132 def to_internal_value(self, value):
133 return IPNetwork(value)
136class SerializedPKRelatedField(PrimaryKeyRelatedField):
137 """
138 Extends PrimaryKeyRelatedField to return a serialized object on read. This is useful for representing related
139 objects in a ManyToManyField while still allowing a set of primary keys to be written.
140 """
141 def __init__(self, serializer, nested=False, **kwargs):
142 self.serializer = serializer
143 self.nested = nested
144 self.pk_field = kwargs.pop('pk_field', None)
146 super().__init__(**kwargs)
148 def to_representation(self, value):
149 return self.serializer(value, nested=self.nested, context={'request': self.context['request']}).data
152@extend_schema_field(OpenApiTypes.INT64)
153class RelatedObjectCountField(serializers.ReadOnlyField):
154 """
155 Represents a read-only integer count of related objects (e.g. the number of racks assigned to a site). This field
156 is detected by get_annotations_for_serializer() when determining the annotations to be added to a queryset
157 depending on the serializer fields selected for inclusion in the response.
158 """
159 def __init__(self, relation, **kwargs):
160 self.relation = relation
162 super().__init__(**kwargs)
165class IntegerRangeSerializer(serializers.Serializer):
166 """
167 Represents a range of integers.
168 """
169 def to_internal_value(self, data):
170 if not isinstance(data, (list, tuple)) or len(data) != 2: 170 ↛ 171line 170 didn't jump to line 171 because the condition on line 170 was never true
171 raise ValidationError(_("Ranges must be specified in the form (lower, upper)."))
172 if type(data[0]) is not int or type(data[1]) is not int: 172 ↛ 173line 172 didn't jump to line 173 because the condition on line 172 was never true
173 raise ValidationError(_("Range boundaries must be defined as integers."))
175 return NumericRange(data[0], data[1] + 1, bounds='[)')
177 def to_representation(self, instance):
178 return get_inclusive_integer_range_bounds(instance)
181class AttributesField(serializers.JSONField):
182 """
183 Custom attributes stored as JSON data.
184 """
185 def to_internal_value(self, data):
186 data = super().to_internal_value(data)
188 # If updating an object, start with the initial attribute data. This enables the client to modify
189 # individual attributes without having to rewrite the entire field.
190 if data and self.parent.instance:
191 initial_data = getattr(self.parent.instance, self.source, None) or {}
192 return {**initial_data, **data}
194 return data