-
Notifications
You must be signed in to change notification settings - Fork 2
Expand file tree
/
Copy pathmodels.py
More file actions
176 lines (144 loc) · 5.37 KB
/
Copy pathmodels.py
File metadata and controls
176 lines (144 loc) · 5.37 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
import swapper
from django.conf import settings
from django.contrib.contenttypes.models import ContentType
from django.db import models
from django.db.models import Count, OuterRef, Subquery, Value
from django.db.models.functions import Coalesce
from django.utils.translation import gettext_lazy as _
from model_utils.models import TimeStampedModel
from baseapp_core.graphql import RelayModel
from baseapp_core.models import (
DocumentIdMixin,
DocumentIdTargetMixin,
DocumentIdUniqueTargetMixin,
)
def default_reports_count():
return {"total": 0}
def default_reports_count_full():
d = default_reports_count()
ReportTypeModel = swapper.load_model("baseapp_reports", "ReportType")
for report_type in ReportTypeModel.objects.all():
d[report_type.key] = 0
return d
class AbstractReportType(DocumentIdMixin, RelayModel, TimeStampedModel):
key = models.CharField(max_length=255, unique=True)
label = models.CharField(max_length=255)
content_types = models.ManyToManyField(
ContentType,
blank=True,
related_name="report_types",
)
parent_type = models.ForeignKey(
swapper.get_model_name("baseapp_reports", "ReportType"),
blank=True,
null=True,
on_delete=models.SET_NULL,
related_name="sub_types",
)
class Meta:
abstract = True
swappable = swapper.swappable_setting("baseapp_reports", "ReportType")
def __str__(self):
return self.label
@classmethod
def get_graphql_object_type(cls):
from .graphql.object_types import ReportTypeObjectType
return ReportTypeObjectType
class AbstractReport(DocumentIdTargetMixin, DocumentIdMixin, RelayModel, TimeStampedModel):
user = models.ForeignKey(
settings.AUTH_USER_MODEL,
related_name="reports",
on_delete=models.CASCADE,
)
report_type = models.ForeignKey(
swapper.get_model_name("baseapp_reports", "ReportType"),
related_name="reports",
on_delete=models.CASCADE,
null=True,
blank=True,
)
report_subject = models.TextField(blank=True, null=True)
class Meta:
abstract = True
swappable = swapper.swappable_setting("baseapp_reports", "Report")
unique_together = [["user", "target_document"]]
def __str__(self):
return "Report (%s) #%s by %s" % (
self.report_type,
self.id,
self.user.first_name,
)
def save(self, *args, **kwargs):
super().save(*args, **kwargs)
self.update_reports_count(self.target)
def delete(self, *args, **kwargs):
target = self.target
super().delete(*args, **kwargs)
self.update_reports_count(target)
@classmethod
def update_reports_count(cls, target):
"""Recompute and persist `reports_count` on `ReportableMetadata` for `target`."""
if not target:
return
ReportableMetadata = swapper.load_model("baseapp_reports", "ReportableMetadata")
metadata = ReportableMetadata.get_or_create_for_object(target)
if metadata is None:
return
counts = default_reports_count_full()
rows = (
cls.objects.filter(
target_document=metadata.target_id,
report_type__isnull=False,
)
.values("report_type__key")
.annotate(n=Count("id"))
)
for row in rows:
key = row["report_type__key"]
n = row["n"]
if key in counts:
counts[key] = n
counts["total"] += n
metadata.reports_count = counts
metadata.save(update_fields=["reports_count", "modified"])
@classmethod
def get_graphql_object_type(cls):
from .graphql.object_types import ReportObjectType
return ReportObjectType
class AbstractReportableMetadata(DocumentIdUniqueTargetMixin, TimeStampedModel):
"""
Stores reporting metadata (per-type counts) for any documentable object.
Linked to `DocumentId` instead of adding columns to each reportable model,
following the plugin architecture pattern for loose coupling.
"""
reports_count = models.JSONField(
default=default_reports_count,
verbose_name=_("reports count"),
editable=False,
)
class Meta:
abstract = True
swappable = swapper.swappable_setting("baseapp_reports", "ReportableMetadata")
verbose_name = _("reportable metadata")
verbose_name_plural = _("reportable metadata")
def __str__(self):
return f"ReportableMetadata for {self.target}"
@classmethod
def annotate_queryset(cls, queryset):
"""
Annotate `queryset` with reportable metadata to prevent N+1 queries when
resolving `reports_count` for many rows of the same model.
Adds `_reportable_reports_count` (zero-total dict when no metadata row exists).
"""
model_cls = queryset.model
ct_id = ContentType.objects.get_for_model(model_cls).pk
metadata_qs = cls.objects.filter(
target__content_type_id=ct_id,
target__object_id=OuterRef("pk"),
)
return queryset.annotate(
_reportable_reports_count=Coalesce(
Subquery(metadata_qs.values("reports_count")[:1]),
Value(default_reports_count(), output_field=models.JSONField()),
),
)