-
Notifications
You must be signed in to change notification settings - Fork 4
Expand file tree
/
Copy pathmodels.py
More file actions
193 lines (158 loc) · 5.04 KB
/
models.py
File metadata and controls
193 lines (158 loc) · 5.04 KB
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
117
118
119
120
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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
from django.db import (
models,
)
import pghistory
import pgtrigger
class CannotDelete(models.Model):
"""This model cannot be deleted.
The ``pgtrigger.Protect`` trigger protects the deletion operation
from happening
"""
class Meta:
triggers = [
pgtrigger.Protect(
name='protect_deletes',
operation=pgtrigger.Delete,
)
]
class AppendOnly(models.Model):
"""This model can only be appended.
The ``pgtrigger.Protect`` trigger protects the update or delete operations
from happening, making this an "append-only" model.
"""
int_field = models.IntegerField()
class Meta:
triggers = [
pgtrigger.Protect(
name='append_only',
operation=(pgtrigger.Update | pgtrigger.Delete),
)
]
class ReadOnlyField(models.Model):
"""
The "created_at" field cannot be updated (i.e. a read-only field).
Updates to other fields will pass, but any updates to created_at will
result in an error
"""
created_at = models.DateTimeField(auto_now_add=True)
int_field = models.IntegerField()
class Meta:
triggers = [
pgtrigger.Protect(
name='read_only_field',
operation=pgtrigger.Update,
condition=pgtrigger.Q(
old__created_at__df=pgtrigger.F("new__created_at")
),
)
]
class SoftDelete(models.Model):
"""
This model cannot be deleted. When a user tries to delete it, the
model will be "soft" deleted instead and have the ``is_active``
boolean set to ``False``
"""
is_active = models.BooleanField(default=True)
class Meta:
triggers = [
pgtrigger.SoftDelete(
name='soft_delete',
field='is_active',
value=False,
)
]
class Versioned(models.Model):
"""
This model is versioned. The "version" field is incremented on every
update, and users cannot directly update the "version" field.
"""
version = models.IntegerField(default=0)
char_field = models.CharField(max_length=32)
class Meta:
triggers = [
pgtrigger.Protect(
name='protect_version_edits',
operation=pgtrigger.Update,
condition=pgtrigger.Q(old__version__df=pgtrigger.F('new__version')),
),
pgtrigger.Trigger(
name='versioned',
when=pgtrigger.Before,
operation=pgtrigger.Update,
func='NEW.version = NEW.version + 1; RETURN NEW;',
condition=pgtrigger.Condition('OLD.* IS DISTINCT FROM NEW.*'),
),
]
class OfficialInterfaceManager(models.Manager):
@pgtrigger.ignore('tutorial.OfficialInterface:protect_inserts')
def official_create(
self,
):
return self.create()
class OfficialInterface(models.Model):
"""
This model has inserts protected and can only be created by
using OfficialInterface.objects.official_create()
"""
objects = OfficialInterfaceManager()
class Meta:
triggers = [
pgtrigger.Protect(
name='protect_inserts',
operation=pgtrigger.Insert,
)
]
class FSM(models.Model):
"""The "status" field can only perform configured transitions during
updates. Any invalid transitions will result in an error.
"""
class Status(models.TextChoices):
UNPUBLISHED = 'unpublished'
PUBLISHED = 'published'
INACTIVE = 'inactive'
status = models.CharField(
choices=Status.choices,
default=Status.UNPUBLISHED,
max_length=16,
)
class Meta:
triggers = [
pgtrigger.FSM(
name='check_status_transitions',
field='status',
transitions=(
(
'unpublished',
'published',
),
(
'unpublished',
'inactive',
),
(
'published',
'inactive',
),
),
)
]
@pghistory.track(
# Create a "snapshot" event on every insert/update
pghistory.Snapshot('snapshot'),
# Create a "create" event whenever a model is created
pghistory.AfterInsert('create'),
# Create a "low_int" event on every update where int_field < 0
pghistory.AfterUpdate(
'low_int',
condition=pgtrigger.Q(new__int_field__lt=0),
),
)
class Tracked(models.Model):
"""
This model uses django-pghistory to track and snapshot field
changes and events on this model. django-pghistory is a library on
top of django-pgtrigger that helps make history tracking on Django
models easy
"""
int_field = models.IntegerField()
char_field = models.CharField(max_length=64)