Skip to content

Instantly share code, notes, and snippets.

@nmanumr
Last active August 13, 2023 23:17
Show Gist options
  • Save nmanumr/527580025929cef936f49acf666c5a84 to your computer and use it in GitHub Desktop.
Save nmanumr/527580025929cef936f49acf666c5a84 to your computer and use it in GitHub Desktop.
from django.db import transaction
from ordered_model.serializers import OrderedModelSerializer
from rest_framework import serializers
from rest_framework.serializers import ListSerializer
class ReplaceListSerializer(ListSerializer):
@transaction.atomic()
def update(self, instances, validated_data):
ins_mapping = {ins.id: ins for ins in instances}
new_obj_ids = [item['id'] for item in validated_data if item.get('id', None) is not None]
# Perform creations and updates.
ret = []
for obj in validated_data:
id = obj.get('id', None)
instance = id and ins_mapping.get(id, None)
if instance is not None:
ret.append(self.child.update(instance, obj))
else:
ret.append(self.child.create({**obj}))
# Perform deletions.
for ins_id, instance in ins_mapping.items():
if ins_id not in new_obj_ids:
instance.delete()
return ret
class ReplaceableListSerializer(serializers.ModelSerializer):
def __init_subclass__(cls, **kwargs):
if 'Meta' in cls.__dict__:
meta_bases = tuple(base_cls.Meta for base_cls in cls.__bases__ if 'Meta' in cls.__dict__)
cls.Meta = type('Meta', (cls.Meta, *meta_bases), {})
@transaction.atomic()
def update(self, instance, validated_data):
replaceable_fields_data = [
{
'data': validated_data.pop(field_name),
'relation_name': relation_name,
'field_name': field_name,
}
for field_name, relation_name in self.get_replaceable_fields()
if field_name in validated_data
]
instance = super().update(instance, validated_data)
for replaceable_field in replaceable_fields_data:
instances = getattr(instance, replaceable_field['field_name']).all()
data = replaceable_field['data']
serializer_cls = self._declared_fields.get(replaceable_field['field_name']).child.__class__
serializer = serializer_cls(data=data, many=True, instance=instances)
serializer.is_valid(raise_exception=True)
serializer.save(**{replaceable_field['relation_name']: instance.pk})
return instance
def get_replaceable_fields(self):
fields = [
(name, getattr(self.Meta.model, name).field.attname)
for name, field_cls in self._declared_fields.items()
if isinstance(field_cls, ReplaceListSerializer)
]
return fields
class Meta:
list_serializer_class = ReplaceListSerializer
class ReplaceableOrderedListSerializer(ReplaceableListSerializer, OrderedModelSerializer):
@transaction.atomic()
def update(self, instance, validated_data):
order = None
order_field = self.get_order_field()
if order_field in validated_data:
order = validated_data.pop(order_field)
instance = super().update(instance, validated_data)
if order is not None:
instance.to(order)
return instance
from rest_framework import serializers
from test_app.models import Subsection, Job, Section
from test_app.replace_list import ReplaceableListSerializer, ReplaceableOrderedListSerializer
class SubsectionSerializer(ReplaceableOrderedListSerializer):
# id needs to be overriden with required=False
# otherwise whole state replacement doesn't work
id = serializers.IntegerField(required=False)
class Meta:
model = Subsection
fields = ("id", "name", "weightage", "order")
class SectionSerializer(ReplaceableOrderedListSerializer):
# id needs to be overriden with required=False
# otherwise whole state replacement doesn't work
id = serializers.IntegerField(required=False)
subsection_set = SubsectionSerializer(many=True, required=False)
class Meta:
model = Section
fields = ("id", "name", "weightage", "order", 'subsection_set')
class JobSerializer(ReplaceableListSerializer):
section_set = SectionSerializer(many=True, required=False)
class Meta:
model = Job
fields = ("id", "name", 'section_set')
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment