Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
import persisting_theory
from rest_framework import serializers
from django.db import models
class ConfNotFound(KeyError):
pass
class Registry(persisting_theory.Registry):
look_into = "mutations"
def connect(self, type, klass, perm_checkers=None):
def decorator(serializer_class):
t = self.setdefault(type, {})
t[klass] = {
"serializer_class": serializer_class,
"perm_checkers": perm_checkers or {},
}
return serializer_class
return decorator
def apply(self, type, obj, payload):
conf = self.get_conf(type, obj)
serializer = conf["serializer_class"](obj, data=payload)
serializer.is_valid(raise_exception=True)
previous_state = serializer.get_previous_state(obj, serializer.validated_data)
serializer.apply(obj, serializer.validated_data)
return previous_state
def is_valid(self, type, obj, payload):
conf = self.get_conf(type, obj)
serializer = conf["serializer_class"](obj, data=payload)
return serializer.is_valid(raise_exception=True)
def get_validated_payload(self, type, obj, payload):
conf = self.get_conf(type, obj)
serializer = conf["serializer_class"](obj, data=payload)
serializer.is_valid(raise_exception=True)
return serializer.payload_serialize(serializer.validated_data)
def has_perm(self, perm, type, obj, actor):
if perm not in ["approve", "suggest"]:
raise ValueError("Invalid permission {}".format(perm))
conf = self.get_conf(type, obj)
checker = conf["perm_checkers"].get(perm)
if not checker:
return False
return checker(obj=obj, actor=actor)
def get_conf(self, type, obj):
try:
type_conf = self[type]
except KeyError:
raise ConfNotFound("{} is not a registered mutation".format(type))
try:
conf = type_conf[obj.__class__]
except KeyError:
try:
conf = type_conf[None]
except KeyError:
raise ConfNotFound(
"No mutation configuration found for {}".format(obj.__class__)
)
return conf
class MutationSerializer(serializers.Serializer):
def apply(self, obj, validated_data):
raise NotImplementedError()
def get_previous_state(self, obj, validated_data):
return
def payload_serialize(self, data):
return data
class UpdateMutationSerializer(serializers.ModelSerializer, MutationSerializer):
serialized_relations = {}
def __init__(self, *args, **kwargs):
# we force partial mode, because update mutations are partial
kwargs.setdefault("partial", True)
super().__init__(*args, **kwargs)
def apply(self, obj, validated_data):
return self.update(obj, validated_data)
def validate(self, validated_data):
if not validated_data:
raise serializers.ValidationError("You must update at least one field")
return super().validate(validated_data)
def db_serialize(self, validated_data):
data = {}
# ensure model fields are serialized properly
for key, value in list(validated_data.items()):
if not isinstance(value, models.Model):
data[key] = value
continue
field = self.serialized_relations[key]
data[key] = getattr(value, field)
return data
def payload_serialize(self, data):
data = super().payload_serialize(data)
# we use our serialized_relations configuration
# to ensure we store ids instead of model instances in our json
# payload
for field, attr in self.serialized_relations.items():
try:
data[field] = getattr(data[field], attr)
except KeyError:
continue
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
return data
def create(self, validated_data):
validated_data = self.db_serialize(validated_data)
return super().create(validated_data)
def get_previous_state(self, obj, validated_data):
return get_update_previous_state(
obj,
*list(validated_data.keys()),
serialized_relations=self.serialized_relations
)
def get_update_previous_state(obj, *fields, serialized_relations={}):
if not fields:
raise ValueError("You need to provide at least one field")
state = {}
for field in fields:
value = getattr(obj, field)
if isinstance(value, models.Model):
# we store the related object id and repr for better UX
id_field = serialized_relations[field]
related_value = getattr(value, id_field)
state[field] = {"value": related_value, "repr": str(value)}
else:
state[field] = {"value": value}
return state
registry = Registry()