approval.models.monitoring

  1import logging
  2from typing import Type, Optional, Iterable, Any
  3from uuid import uuid4
  4
  5from annoying.fields import AutoOneToOneField
  6from django.conf import settings
  7from django.contrib.auth.models import AbstractUser
  8from django.core.exceptions import ValidationError
  9from django.core.serializers.json import DjangoJSONEncoder
 10from django.db import models
 11from django.db.models.base import ModelBase
 12from django.utils import timezone
 13from django.utils.functional import cached_property
 14from django.utils.translation import gettext_lazy as _, pgettext_lazy
 15
 16from approval.models import MonitoredModel
 17from approval.signals import pre_approval, post_approval
 18
 19logger = logging.getLogger("approval")
 20
 21
 22class SandboxMeta(ModelBase):
 23    """Metaclass to create a dynamic sandbox model."""
 24
 25    def __new__(cls, name: str, bases: tuple, attrs: dict, **kwargs) -> Type:
 26        # Names
 27        source_model: Type[models.Model] = attrs.get("base", None)
 28        source_name: str = f"{source_model._meta.model_name}"
 29        source_app: str = f"{source_model._meta.app_label}"
 30        source_fqmn: str = f"{source_model._meta.app_label}.{source_model._meta.model_name}"
 31        reverse_name: str = f"moderated_{source_model._meta.model_name.lower()}_approval"
 32        table_name: str = f"{source_model._meta.db_table}_approval"
 33        source_verbose_name: str = source_model._meta.verbose_name
 34        permission_names: dict = {"moderate": f"moderate_{table_name}"}
 35
 36        # Dynamic class replacing the original class.
 37        class DynamicSandbox(models.Model):
 38            """
 39            Base class model to monitor changes on a source Model.
 40
 41            Notes:
 42                To define a model holding changes detected on a source model,
 43                the developer must declare a model class inheriting from
 44                `Approval`. For it to function properly, the developer must
 45                provide some configuration in the form of 5 attributes.
 46            """
 47            _is_sandbox: bool = True  # Attribute to detect dynamic class as sandbox
 48            MODERATION_STATUS: tuple = (
 49                (None, pgettext_lazy("approval.moderation", "Pending")),
 50                (False, pgettext_lazy("approval.moderation", "Rejected")),
 51                (True, pgettext_lazy("approval.moderation", "Approved")),
 52            )
 53            DRAFT_STATUS: tuple = (
 54                (False, pgettext_lazy("approval.draft", "Draft")),
 55                (True, pgettext_lazy("approval.draft", "Waiting for moderation")),
 56            )
 57
 58            base: Type[models.Model] = attrs.get("base", None)
 59            approval_fields: list[str] = attrs.get("approval_fields", [])
 60            approval_store_fields: list[str] = attrs.get("approval_store_fields", [])
 61            approval_default: dict[str, object] = attrs.get("approval_default", {})
 62            auto_approve_staff: bool = attrs.get("auto_approve_staff", True)
 63            auto_approve_new: bool = attrs.get("auto_approve_new", False)
 64            auto_approve_by_request: bool = attrs.get("auto_approve_by_request", True)
 65            delete_on_approval: bool = attrs.get("delete_on_approval", False)
 66
 67            uuid = models.UUIDField(default=uuid4, verbose_name=_("UUID"))
 68            source = models.OneToOneField(
 69                source_fqmn, null=False, on_delete=models.CASCADE, related_name="approval"
 70            )
 71            sandbox = models.JSONField(
 72                default=dict,
 73                blank=False,
 74                encoder=DjangoJSONEncoder,
 75                verbose_name=pgettext_lazy("approval_entry", "Data"),
 76            )
 77            approved = models.BooleanField(
 78                default=None,
 79                null=True,
 80                choices=MODERATION_STATUS,
 81                verbose_name=pgettext_lazy("approval_entry", "Moderated"),
 82            )
 83            moderator = models.ForeignKey(
 84                settings.AUTH_USER_MODEL,
 85                default=None,
 86                blank=True,
 87                null=True,
 88                related_name=reverse_name,
 89                verbose_name=pgettext_lazy("approval_entry", "Moderated by"),
 90                on_delete=models.CASCADE,
 91            )
 92            draft = models.BooleanField(
 93                default=True,
 94                choices=DRAFT_STATUS,
 95                verbose_name=pgettext_lazy("approval_entry", "Draft"),
 96            )
 97            approval_date = models.DateTimeField(
 98                null=True, verbose_name=pgettext_lazy("approval_entry", "Moderated at")
 99            )
100            info = models.TextField(blank=True, verbose_name=pgettext_lazy("approval", "Reason"))
101            updated = models.DateTimeField(
102                auto_now=True, verbose_name=pgettext_lazy("approval_entry", "Updated")
103            )
104
105            class Meta:
106                abstract = False
107                db_table = table_name
108                app_label = source_app
109                verbose_name = pgettext_lazy("approval", "{name} approval").format(
110                    name=source_verbose_name
111                )
112                verbose_name_plural = pgettext_lazy("approval", "{name} approvals").format(
113                    name=source_verbose_name
114                )
115                permissions = [[permission_names["moderate"], f"can moderate {source_name}"]]
116
117            def save(self, **options):
118                return super().save(**options)
119
120            def __str__(self):
121                return f"{self.source} approval status: {self.get_approved_display()}"
122
123            def __repr__(self):
124                return f"{self._meta.model_name}(uuid={self.uuid}, ...)"
125
126            # API
127            def _update_source(self, default: bool = False, save: bool = False) -> bool:
128                """
129                Validate pending changes and apply them to source instance.
130
131                Keyword Args:
132                    default:
133                        If `True`, reset source fields with their default values, as
134                        specified by the attribute `ApprovalModel.approval_default`.
135                    save:
136                        If `True`, the source instance is saved in the database.
137
138                Returns:
139                    `True` if source could be updated with no issue, meaning data set into
140                    the fields in source is correct (correct type and values).
141                    `False` if the data put into the source fields can not be validated
142                    by Django.
143                """
144                original = dict()  # Revert to target values in case of failure
145                if default is False:
146                    for field in self._get_fields_data():
147                        original[field] = getattr(self.source, field)
148                        setattr(self.source, field, self.sandbox["fields"][field])
149                    for field in self._get_store_fields_data():
150                        original[field] = getattr(self.source, field)
151                        setattr(self.source, field, self.sandbox["store"][field])
152                    logger.debug(
153                        pgettext_lazy("approval", f"Updated monitored object fields using sandbox.")
154                    )
155                else:
156                    for field in self.approval_default.keys():
157                        setattr(self.source, field, self.approval_default[field])
158                    logger.debug(
159                        pgettext_lazy(
160                            "approval", f"Updated monitored object fields using default values."
161                        )
162                    )
163
164                try:
165                    self.source.clean_fields()  # use Django field validation on new values
166                    if save:
167                        self.source._ignore_approval = True
168                        self.source.save()
169                        logger.debug(
170                            pgettext_lazy("approval", f"Updated monitored object was persisted.")
171                        )
172                    return True
173                except ValidationError as exc:
174                    for key in exc.message_dict:
175                        logger.debug(
176                            pgettext_lazy(
177                                "approval", "Field {name} could not be persisted to model."
178                            ).format(name=key)
179                        )
180                        if hasattr(self.source, key):
181                            setattr(self.source, key, original[key])
182                    if save:
183                        self.source._ignore_approval = True
184                        self.source.save()
185                    return False
186
187            def _update_sandbox(self, slot: str = None, source: MonitoredModel = None) -> None:
188                """
189                Copy source monitored field values into the sandbox.
190
191                Keyword Args:
192                    slot:
193                        Field values can be saved in various slots of
194                        the same approval instance. Slots are arbitrary names.
195                        Default slot is `None`.
196                    source:
197                        Can be a reference of another object of the same model
198                        as `self.source`, for example to use it as a template.
199                """
200                slot: str = slot or "fields"
201                source: MonitoredModel = source or self.source
202                # Save monitored fields into the defined slot.
203                fields: list = self._get_field_names()
204                values: dict[str, Any] = {
205                    k: getattr(source, k) for k in fields if hasattr(source, k)
206                }
207                self.sandbox[slot] = values
208                # Save store/restore fields into the "store" slot.
209                fields: list[str] = self._get_store_field_names()
210                values: dict[str, Any] = {
211                    k: getattr(source, k) for k in fields if hasattr(source, k)
212                }
213                self.sandbox["store"] = values
214                # Mark the changes as Pending
215                self.approved = None
216                self.save()
217                logger.debug(
218                    pgettext_lazy("approval", "Sandbox for {source} was updated.").format(
219                        source=source
220                    )
221                )
222
223            def _needs_approval(self) -> bool:
224                """
225                Get whether the status of the object really needs an approval.
226
227                Returns:
228                    `True` if the status of the object can be auto-approved, `False` otherwise.
229
230                In this case, the diff is `None` when the target object was updated
231                but none of the monitored fields was changed.
232                """
233                return self._get_diff() is not None
234
235            @cached_property
236            def _get_valid_fields(self) -> Optional[set[str]]:
237                """
238                Return the names of the data fields that can be used to update the source.
239
240                Returns:
241                    A list of field names that are relevant to the source (useful
242                    when the original model changed after a migration).
243
244                Without this method, applying values from the sandbox would fail
245                if a monitored field was removed from the model.
246                """
247                fields = set(self._get_field_names())
248                source_fields = set(self.source._meta.get_all_field_names())
249                return fields.intersection(source_fields) or None
250
251            def _get_field_names(self) -> list[str]:
252                """Get the list of monitored field names."""
253                return self.approval_fields
254
255            def _get_store_field_names(self) -> list[str]:
256                return self.approval_store_fields
257
258            def _get_fields_data(self) -> dict[str, Any]:
259                """
260                Return a dictionary of the data in the sandbox.
261
262                Returns:
263                    A dictionary with field names as keys and their sandbox value as values.
264                """
265                return self.sandbox.get("fields", {})
266
267            def _get_store_fields_data(self) -> dict[str, Any]:
268                """
269                Return a dictionary of the data in the sandbox store.
270
271                Returns:
272                    A dictionary with field names as keys and their sandbox value as values.
273                """
274                return self.sandbox.get("store", {})
275
276            def _get_diff(self) -> Optional[list[str]]:
277                """
278                Get the fields that have been changed from source to sandbox.
279
280                Returns:
281                    A list of monitored field names that are different in the source.
282                    `None` if no difference exists between the source and the sandbox.
283                """
284                data = self._get_fields_data()
285                source_data = {
286                    field: getattr(self.source, field) for field in self._get_field_names()
287                }
288                return [key for key in data.keys() if data[key] != source_data[key]] or None
289
290            def _can_user_bypass_approval(self, user: AbstractUser) -> bool:
291                """
292                Get whether a user can bypass approval rights control.
293
294                Args:
295                    user:
296                        The user instance to check against.
297                """
298                permission_name: str = f"{self.base._meta.app_label}.{permission_names['moderate']}"
299                return user and user.has_perm(perm=permission_name)
300
301            def _auto_process_approval(
302                self, authors: Iterable = None, update: bool = False
303            ) -> None:
304                """
305                Approve or deny edits automatically.
306
307                This method denies or approves according to configuration of
308                "auto_approve_..." fields.
309
310                Args:
311                    authors:
312                        The list of users responsible for the change. If the instance
313                        contains a `request` attribute, the connected user is considered the
314                        author of the change.
315                    update:
316                        Is used to differentiate between new objects and updated ones.
317                        Is set to `False` when the object is new, `True` otherwise.
318                """
319                authorized: bool = any(self._can_user_bypass_approval(author) for author in authors)
320                optional: bool = not self._needs_approval()
321
322                if authorized or optional or (self.auto_approve_new and not update):
323                    self.approve(user=authors[0], save=True)
324                if self.auto_approve_staff and any((u.is_staff for u in authors)):
325                    self.approve(user=authors[0], save=True)
326                self.auto_process_approval(authors=authors)
327
328            def _get_authors(self) -> Iterable:
329                """
330                Get the authors of the source instance.
331
332                Warnings:
333                    This method *must* be overriden in the concrete model.
334                """
335                raise NotImplemented("You must define _get_authors() in your model.")
336
337            def submit_approval(self) -> bool:
338                """
339                Sets the status of the object to Waiting for moderation.
340
341                In other words, the monitored object will get moderated only
342                after it's pulled from draft.
343                """
344                if self.draft:
345                    self.draft = False
346                    self.save()
347                    logger.debug(
348                        pgettext_lazy("approval", f"Set sandbox as waiting for moderation.")
349                    )
350                    return True
351                return False
352
353            def is_draft(self) -> bool:
354                """Check whether the object is currently in draft mode."""
355                return self.draft
356
357            def approve(self, user=None, save: bool = False) -> None:
358                """
359                Approve pending edits.
360
361                Args:
362                    user:
363                        Instance of user who moderated the content.
364                    save:
365                        If `True`, persist changes to the monitored object.
366                        If `False`, apply sandbox to the monitored object, but don't save it.
367                """
368                pre_approval.send(self.base, instance=self.source, status=self.approved)
369                self.approval_date = timezone.now()
370                self.approved = True
371                self.moderator = user
372                self.draft = False
373                self.info = pgettext_lazy(
374                    "approval_entry", "Congratulations, your edits have been approved."
375                )
376                self._update_source(save=True)  # apply changes to monitored object
377                if save:
378                    super().save()
379                # If the approval instance is to be deleted upon approval, do it here
380                if self.delete_on_approval:
381                    super().delete()
382                    logger.debug(pgettext_lazy("approval", f"Changes in sandbox were deleted upon approval."))
383                post_approval.send(self.base, instance=self.source, status=self.approved)
384                logger.debug(pgettext_lazy("approval", f"Changes in sandbox were approved."))
385
386            def deny(self, user=None, reason: str = None, save: bool = False) -> None:
387                """
388                Reject pending edits.
389
390                Args:
391                    user:
392                        Instance of user who moderated the content.
393                    reason:
394                        String explaining why the content was refused.
395                    save:
396                        If `True`, persist changes to the monitored object.
397                        If `False`, apply sandbox to the monitored object, but don't save it.
398                """
399                pre_approval.send(self.base, instance=self.source, status=self.approved)
400                self.moderator = user
401                self.approved = False
402                self.draft = False
403                self.info = reason or pgettext_lazy(
404                    "approval_entry", "Your edits have been refused."
405                )
406                if save:
407                    self.save()
408                post_approval.send(self.base, instance=self.source, status=self.approved)
409                logger.debug(pgettext_lazy("approval", f"Changes in sandbox were rejected."))
410
411            def auto_process_approval(self, authors: Iterable = None) -> None:
412                """
413                User-defined auto-processing, the developer should override this.
414
415                Auto-processing is the choice of action regarding the author
416                or the state of the changes. This method can choose to auto-approve
417                for some users or auto-deny changes from inappropriate IPs.
418
419                """
420                return None
421
422        # Patch the dynamic sandbox methods with those provided by the original class
423        for attr in attrs:
424            if callable(attrs.get(attr)):
425                setattr(DynamicSandbox, attr, attrs.get(attr))
426        return DynamicSandbox
427
428
429class Sandbox:
430    """
431    Class providing attributes to configure a Sandbox model.
432
433    To use this class, you need to create a model class that inherits from
434    this class, and use `SandboxBase` as a metaclass:
435
436    ```python
437    class EntryApproval(SandboxModel, metaclass=SandboxBase):
438        base = Entry
439        approval_fields = ["description", "content"]
440        approval_store_fields = ["is_visible"]
441        approval_default = {"is_visible": False, "description": ""}
442        auto_approve_staff = False
443        auto_approve_new = False
444        auto_approve_by_request = False
445
446        def _get_authors(self) -> Iterable:
447            return [self.source.user]
448    ```
449
450    Attributes:
451        base:
452            Monitored model class.
453        approval_fields (list[str]):
454            List of model field names on the base model that should
455            be monitored and should trigger the approval process.
456        approval_default:
457            When a new object is created and immediately needs approval,
458            define the default values for the source while waiting for
459            approval. For example, for a blog entry, you can set the default `published`
460            attribute to `False`.
461        approval_store_fields:
462            List of model field names that should be stored in the approval
463            state, even though the field is not monitored. Those fields will
464            be restored to the object when approved. Generally contains
465            fields used in approval_default.
466        auto_approve_staff:
467            If `True`, changes made by a staff member should be applied
468            immediately, bypassing moderation.
469        auto_approve_new:
470            If `True`, a new object created would bypass the approval
471            phase and be immediately persisted.
472        auto_approve_by_request:
473            If `True` the user in the object's request attribute, if any,
474            is used to test if the object can be automatically approved.
475            If `False`, use the default object author only.
476        delete_on_approval:
477            If `True`, the approval metadata will be deleted upon approval.
478            `False` by default, since you probably want to see who accepted
479            an entry and when.
480    """
481
482    base: Type[models.Model] = None
483    approval_fields: list[str] = []
484    approval_store_fields: list[str] = []
485    approval_default: dict[str, object] = {}
486    auto_approve_staff: bool = True
487    auto_approve_new: bool = False
488    auto_approve_by_request: bool = True
489    delete_on_approval: bool = False