Coverage for django_napse/utils/serializers/serializer.py: 100%
125 statements
« prev ^ index » next coverage.py v7.4.3, created at 2024-03-12 13:49 +0000
1from __future__ import annotations
2from django_napse.utils.serializers.fields import Field
3import operator
4from rest_framework.serializers import ValidationError
7class BaseSerializer:
8 _fields = {}
9 _compiled_fields = {}
10 _validators: dict[str, tuple[bool, callable]] = {}
13class MetaSerializer(type):
14 @staticmethod
15 def _compile_fields(fields, serializer_cls: Serializer) -> list[tuple]:
16 """Compile a field to give an easy access to all elements of each fields."""
18 def _compile_field(field, name, serializer_cls: Serializer) -> tuple:
19 """Compile field's elements to tuple."""
20 if field.source is not None and "." in field.source:
21 getter = (operator.attrgetter(attr) for attr in field.source.split("."))
22 getter_is_generator = True
23 else:
24 getter = field.as_getter(name, serializer_cls) or operator.attrgetter(name)
25 getter_is_generator = False
26 return (
27 name,
28 field.to_value,
29 field.required,
30 getter,
31 field.getter_takes_serializer,
32 getter_is_generator,
33 )
35 return [_compile_field(field, name, serializer_cls) for name, field in fields.items()]
37 def __new__(
38 cls: MetaSerializer,
39 name: str,
40 bases: tuple,
41 attrs: dict,
42 ):
43 """Define how to build Serializer classes.
45 Args:
46 cls: MetaClass class.
47 name: Class name.
48 bases: Bases of the class.
49 attrs: Method and attributs of the class (like Fields).
50 """
51 fields = {}
53 # Get direct Fields from attributs
54 fields = {key: value for key, value in attrs.items() if isinstance(value, Field)}
55 for key in fields:
56 del attrs[key]
58 # Make serializer class
59 serializer_cls = super(MetaSerializer, cls).__new__(cls, name, bases, attrs)
61 # Retrieve Fields from parent classes
62 for obj in serializer_cls.__mro__[::-1]:
63 if issubclass(obj, BaseSerializer):
64 fields.update(obj._fields)
66 serializer_cls._fields = fields
67 serializer_cls._compiled_fields = cls._compile_fields(fields, serializer_cls)
68 serializer_cls._validators = {
69 name: (
70 field.required,
71 field.validate,
72 )
73 for name, field in fields.items()
74 if field.validate
75 }
77 return serializer_cls
80class Serializer(BaseSerializer, Field, metaclass=MetaSerializer):
81 Model = None
83 def __init__(
84 self,
85 instance=None,
86 data=None,
87 many=False,
88 **kwargs,
89 ):
90 self._instance = instance
91 self._data = data
92 self._many = many
93 self._validated_data: dict[str, any] | None = None
95 if self._data is not None:
96 self.validate_data(self._data)
98 # To use serializer in serializer
99 super().__init__(**kwargs)
101 @property
102 def data(self):
103 if self._data is not None:
104 return self._data
105 return self.to_value()
107 @property
108 def validated_data(self):
109 if self._validated_data is None:
110 error_msg: str = "Data are invalid"
111 raise ValueError(error_msg)
112 return self._validated_data
114 def to_value(self, instance: any | None = None) -> any:
115 """Serialize instance."""
116 instance = instance or self._instance
117 if self._many:
118 return [self._serialize(ist, self._compiled_fields) for ist in instance]
119 return self._serialize(instance, self._compiled_fields)
121 def _serialize(self, instance, fields):
122 serialized_instance = {}
123 for name, to_value, required, getter, getter_takes_serializer, getter_is_generator in fields:
124 try:
125 if getter_is_generator:
126 value = instance
127 for get in getter:
128 value = get(value)
129 else:
130 value = getter(self, instance) if getter_takes_serializer else getter(instance)
131 except (AttributeError, KeyError):
132 if required:
133 error_msg: str = f"Field {name} is required."
134 raise ValueError(error_msg) from None
135 continue
136 if to_value:
137 value = to_value(value)
138 serialized_instance[name] = value
139 return serialized_instance
141 def validate_data(self, data):
142 if not isinstance(data, dict):
143 error_msg: str = "Data must be a dictionary."
144 raise ValidationError(error_msg)
145 for name, (required, validator) in self._validators.items():
146 elt = data.get(name)
147 if elt is None:
148 if required:
149 error_msg: str = f"Field {name} is required."
150 raise ValidationError(error_msg)
151 continue
153 result = validator(elt)
154 if not result:
155 error_msg: str = f"Field {name} is invalid."
156 raise ValidationError(error_msg)
158 # Retrieve db instances if needed
159 for name, field in self._fields.items():
160 if isinstance(field, Serializer):
161 data[name] = field.get(data[name])
163 self._validated_data = data
164 return data
166 def validate(self, data):
167 """Used for automatic validation with serializer.
169 Please to not use this method.
170 Note: provided data must be id or uuid
171 """
172 instance = self.get(data)
173 data = self.to_value(instance)
174 return self.validate_data(data)
176 def _model_checks(self, validated_data):
177 if not self.Model:
178 error_msg: str = "Model is not defined."
179 raise ValueError(error_msg)
181 if not validated_data:
182 error_msg: str = "Data are not validated."
183 raise ValueError(error_msg)
185 def get(self, data):
186 """Retrieve instance from data."""
187 try:
188 instance = self.Model.objects.get(uuid=data) if hasattr(self.Model, "uuid") else self.Model.objects.get(id=data)
189 except (self.Model.DoesNotExist, TypeError):
190 error_msg: str = f"Instance with id {data} does not exist."
191 raise ValidationError(error_msg) from None
192 return instance
194 def create(self, validated_data=None):
195 validated_data = validated_data or self._validated_data
196 self._model_checks(validated_data=validated_data)
197 return self.Model.objects.create(**validated_data)
199 def update(self, instance, validated_data=None):
200 validated_data = validated_data or self._validated_data
201 self._model_checks(validated_data=validated_data)
202 for key, value in validated_data.items():
203 setattr(instance, key, value)
204 instance.save()
205 return instance