See #228: serializer logic

api/funkwhale_api/common/serializers.py

+from rest_framework import serializers
+
+
+class ActionSerializer(serializers.Serializer):
+    """
+    A special serializer that can operate on a list of objects
+    and apply actions on it.
+    """
+
+    action = serializers.CharField(required=True)
+    objects = serializers.JSONField(required=True)
+    filters = serializers.DictField(required=False)
+    actions = None
+    filterset_class = None
+
+    def __init__(self, *args, **kwargs):
+        self.queryset = kwargs.pop('queryset')
+        if self.actions is None:
+            raise ValueError(
+                'You must declare a list of actions on '
+                'the serializer class')
+
+        for action in self.actions:
+            handler_name = 'handle_{}'.format(action)
+            assert hasattr(self, handler_name), (
+                '{} miss a {} method'.format(
+                    self.__class__.__name__, handler_name)
+            )
+        super().__init__(self, *args, **kwargs)
+
+    def validate_action(self, value):
+        if value not in self.actions:
+            raise serializers.ValidationError(
+                '{} is not a valid action. Pick one of {}.'.format(
+                    value, ', '.join(self.actions)
+                )
+            )
+        return value
+
+    def validate_objects(self, value):
+        qs = None
+        if value == 'all':
+            return self.queryset.all().order_by('id')
+        if type(value) in [list, tuple]:
+            return self.queryset.filter(pk__in=value).order_by('id')
+
+        raise serializers.ValidationError(
+            '{} is not a valid value for objects. You must provide either a '
+            'list of identifiers or the string "all".'.format(value))
+
+    def validate(self, data):
+        if not self.filterset_class or 'filters' not in data:
+            # no additional filters to apply, we just skip
+            return data
+
+        qs_filterset = self.filterset_class(
+            data['filters'], queryset=data['objects'])
+        try:
+            assert qs_filterset.form.is_valid()
+        except (AssertionError, TypeError):
+            raise serializers.ValidationError('Invalid filters')
+        data['objects'] = qs_filterset.qs
+        return data
+
+    def save(self):
+        handler_name = 'handle_{}'.format(self.validated_data['action'])
+        handler = getattr(self, handler_name)
+        result = handler(self.validated_data['objects'])
+        payload = {
+            'updated': self.validated_data['objects'].count(),
+            'action': self.validated_data['action'],
+            'result': result,
+        }
+        return payload

api/tests/common/test_serializers.py

+import django_filters
+
+from funkwhale_api.common import serializers
+from funkwhale_api.users import models
+
+
+class TestActionFilterSet(django_filters.FilterSet):
+    class Meta:
+        model = models.User
+        fields = ['is_active']
+
+
+class TestSerializer(serializers.ActionSerializer):
+    actions = ['test']
+    filterset_class = TestActionFilterSet
+
+    def handle_test(self, objects):
+        return {'hello': 'world'}
+
+
+def test_action_serializer_validates_action():
+    data = {'objects': 'all', 'action': 'nope'}
+    serializer = TestSerializer(data, queryset=models.User.objects.none())
+
+    assert serializer.is_valid() is False
+    assert 'action' in serializer.errors
+
+
+def test_action_serializer_validates_objects():
+    data = {'objects': 'nope', 'action': 'test'}
+    serializer = TestSerializer(data, queryset=models.User.objects.none())
+
+    assert serializer.is_valid() is False
+    assert 'objects' in serializer.errors
+
+
+def test_action_serializers_objects_clean_ids(factories):
+    user1 = factories['users.User']()
+    user2 = factories['users.User']()
+
+    data = {'objects': [user1.pk], 'action': 'test'}
+    serializer = TestSerializer(data, queryset=models.User.objects.all())
+
+    assert serializer.is_valid() is True
+    assert list(serializer.validated_data['objects']) == [user1]
+
+
+def test_action_serializers_objects_clean_all(factories):
+    user1 = factories['users.User']()
+    user2 = factories['users.User']()
+
+    data = {'objects': 'all', 'action': 'test'}
+    serializer = TestSerializer(data, queryset=models.User.objects.all())
+
+    assert serializer.is_valid() is True
+    assert list(serializer.validated_data['objects']) == [user1, user2]
+
+
+def test_action_serializers_save(factories, mocker):
+    handler = mocker.spy(TestSerializer, 'handle_test')
+    user1 = factories['users.User']()
+    user2 = factories['users.User']()
+
+    data = {'objects': 'all', 'action': 'test'}
+    serializer = TestSerializer(data, queryset=models.User.objects.all())
+
+    assert serializer.is_valid() is True
+    result = serializer.save()
+    assert result == {
+        'updated': 2,
+        'action': 'test',
+        'result': {'hello': 'world'},
+    }
+    handler.assert_called_once()
+
+
+def test_action_serializers_filterset(factories):
+    user1 = factories['users.User'](is_active=False)
+    user2 = factories['users.User'](is_active=True)
+
+    data = {
+        'objects': 'all',
+        'action': 'test',
+        'filters': {'is_active': True},
+    }
+    serializer = TestSerializer(data, queryset=models.User.objects.all())
+
+    assert serializer.is_valid() is True
+    assert list(serializer.validated_data['objects']) == [user2]