diff --git a/wagtail/core/models/__init__.py b/wagtail/core/models/__init__.py index d5fd112765..6b07439ed3 100644 --- a/wagtail/core/models/__init__.py +++ b/wagtail/core/models/__init__.py @@ -9,7 +9,6 @@ from urllib.parse import urlparse from django import forms from django.apps import apps from django.conf import settings -from django.contrib.auth import get_user_model from django.contrib.auth.models import Group from django.contrib.contenttypes.models import ContentType from django.core import checks @@ -52,6 +51,7 @@ from wagtail.core.utils import ( get_supported_content_language_variant, resolve_model_string) from wagtail.search import index +from .audit_log import BaseLogEntry, BaseLogEntryManager, LogEntryQuerySet # noqa from .collections import ( # noqa BaseCollectionManager, Collection, CollectionManager, CollectionMember, CollectionViewRestriction, GroupCollectionPermission, GroupCollectionPermissionManager, @@ -4316,65 +4316,6 @@ class TaskState(models.Model): verbose_name_plural = _('Task states') -class LogEntryQuerySet(models.QuerySet): - def get_users(self): - """ - Returns a QuerySet of Users who have created at least one log entry in this QuerySet. - - The returned queryset is ordered by the username. - """ - User = get_user_model() - return User.objects.filter( - pk__in=set(self.values_list('user__pk', flat=True)) - ).order_by(User.USERNAME_FIELD) - - -class BaseLogEntryManager(models.Manager): - def get_queryset(self): - return LogEntryQuerySet(self.model, using=self._db) - - def get_instance_title(self, instance): - return str(instance) - - def log_action(self, instance, action, **kwargs): - """ - :param instance: The model instance we are logging an action for - :param action: The action. Should be namespaced to app (e.g. wagtail.create, wagtail.workflow.start) - :param kwargs: Addition fields to for the model deriving from BaseLogEntry - - user: The user performing the action - - title: the instance title - - data: any additional metadata - - content_changed, deleted - Boolean flags - :return: The new log entry - """ - data = kwargs.pop('data', '') - title = kwargs.pop('title', None) - if not title: - title = self.get_instance_title(instance) - - timestamp = kwargs.pop('timestamp', timezone.now()) - return self.model.objects.create( - content_type=ContentType.objects.get_for_model(instance, for_concrete_model=False), - label=title, - action=action, - timestamp=timestamp, - data_json=json.dumps(data), - **kwargs, - ) - - def get_for_model(self, model): - # Return empty queryset if the given object is not valid. - if not issubclass(model, models.Model): - return self.none() - - ct = ContentType.objects.get_for_model(model) - - return self.filter(content_type=ct) - - def get_for_user(self, user_id): - return self.filter(user=user_id) - - class PageLogEntryManager(BaseLogEntryManager): def get_instance_title(self, instance): @@ -4385,113 +4326,6 @@ class PageLogEntryManager(BaseLogEntryManager): return super().log_action(instance, action, **kwargs) -class BaseLogEntry(models.Model): - content_type = models.ForeignKey( - ContentType, - models.SET_NULL, - verbose_name=_('content type'), - blank=True, null=True, - related_name='+', - ) - label = models.TextField() - - action = models.CharField(max_length=255, blank=True, db_index=True) - data_json = models.TextField(blank=True) - timestamp = models.DateTimeField(verbose_name=_('timestamp (UTC)')) - - user = models.ForeignKey( - settings.AUTH_USER_MODEL, - null=True, # Null if actioned by system - blank=True, - on_delete=models.DO_NOTHING, - db_constraint=False, - related_name='+', - ) - - # Flags for additional context to the 'action' made by the user (or system). - content_changed = models.BooleanField(default=False, db_index=True) - deleted = models.BooleanField(default=False) - - objects = BaseLogEntryManager() - - action_registry = None - - class Meta: - abstract = True - verbose_name = _('log entry') - verbose_name_plural = _('log entries') - ordering = ['-timestamp'] - - def save(self, *args, **kwargs): - self.full_clean() - return super().save(*args, **kwargs) - - def clean(self): - self.action_registry.scan_for_actions() - - if self.action not in self.action_registry.actions: - raise ValidationError({'action': _("The log action '{}' has not been registered.").format(self.action)}) - - def __str__(self): - return "LogEntry %d: '%s' on '%s'" % ( - self.pk, self.action, self.object_verbose_name() - ) - - @cached_property - def user_display_name(self): - """ - Returns the display name of the associated user; - get_full_name if available and non-empty, otherwise get_username. - Defaults to 'system' when none is provided - """ - if self.user_id: - try: - user = self.user - except self._meta.get_field('user').related_model.DoesNotExist: - # User has been deleted - return _('user %(id)d (deleted)') % {'id': self.user_id} - - try: - full_name = user.get_full_name().strip() - except AttributeError: - full_name = '' - return full_name or user.get_username() - - else: - return _('system') - - @cached_property - def data(self): - """ - Provides deserialized data - """ - if self.data_json: - return json.loads(self.data_json) - else: - return {} - - @cached_property - def object_verbose_name(self): - model_class = self.content_type.model_class() - if model_class is None: - return self.content_type_id - - return model_class._meta.verbose_name.title - - def object_id(self): - raise NotImplementedError - - def format_message(self): - return self.action_registry.format_message(self) - - def format_comment(self): - return self.action_registry.format_comment(self) - - @property - def comment(self): - return self.format_comment() - - class PageLogEntry(BaseLogEntry): page = models.ForeignKey( 'wagtailcore.Page', diff --git a/wagtail/core/models/audit_log.py b/wagtail/core/models/audit_log.py new file mode 100644 index 0000000000..0dd2d986fc --- /dev/null +++ b/wagtail/core/models/audit_log.py @@ -0,0 +1,176 @@ +import json + +from django.conf import settings +from django.contrib.auth import get_user_model +from django.contrib.contenttypes.models import ContentType +from django.core.exceptions import ValidationError +from django.db import models +from django.utils import timezone +from django.utils.functional import cached_property +from django.utils.translation import gettext_lazy as _ + + +class LogEntryQuerySet(models.QuerySet): + def get_users(self): + """ + Returns a QuerySet of Users who have created at least one log entry in this QuerySet. + + The returned queryset is ordered by the username. + """ + User = get_user_model() + return User.objects.filter( + pk__in=set(self.values_list('user__pk', flat=True)) + ).order_by(User.USERNAME_FIELD) + + +class BaseLogEntryManager(models.Manager): + def get_queryset(self): + return LogEntryQuerySet(self.model, using=self._db) + + def get_instance_title(self, instance): + return str(instance) + + def log_action(self, instance, action, **kwargs): + """ + :param instance: The model instance we are logging an action for + :param action: The action. Should be namespaced to app (e.g. wagtail.create, wagtail.workflow.start) + :param kwargs: Addition fields to for the model deriving from BaseLogEntry + - user: The user performing the action + - title: the instance title + - data: any additional metadata + - content_changed, deleted - Boolean flags + :return: The new log entry + """ + data = kwargs.pop('data', '') + title = kwargs.pop('title', None) + if not title: + title = self.get_instance_title(instance) + + timestamp = kwargs.pop('timestamp', timezone.now()) + return self.model.objects.create( + content_type=ContentType.objects.get_for_model(instance, for_concrete_model=False), + label=title, + action=action, + timestamp=timestamp, + data_json=json.dumps(data), + **kwargs, + ) + + def get_for_model(self, model): + # Return empty queryset if the given object is not valid. + if not issubclass(model, models.Model): + return self.none() + + ct = ContentType.objects.get_for_model(model) + + return self.filter(content_type=ct) + + def get_for_user(self, user_id): + return self.filter(user=user_id) + + +class BaseLogEntry(models.Model): + content_type = models.ForeignKey( + ContentType, + models.SET_NULL, + verbose_name=_('content type'), + blank=True, null=True, + related_name='+', + ) + label = models.TextField() + + action = models.CharField(max_length=255, blank=True, db_index=True) + data_json = models.TextField(blank=True) + timestamp = models.DateTimeField(verbose_name=_('timestamp (UTC)')) + + user = models.ForeignKey( + settings.AUTH_USER_MODEL, + null=True, # Null if actioned by system + blank=True, + on_delete=models.DO_NOTHING, + db_constraint=False, + related_name='+', + ) + + # Flags for additional context to the 'action' made by the user (or system). + content_changed = models.BooleanField(default=False, db_index=True) + deleted = models.BooleanField(default=False) + + objects = BaseLogEntryManager() + + action_registry = None + + class Meta: + abstract = True + verbose_name = _('log entry') + verbose_name_plural = _('log entries') + ordering = ['-timestamp'] + + def save(self, *args, **kwargs): + self.full_clean() + return super().save(*args, **kwargs) + + def clean(self): + self.action_registry.scan_for_actions() + + if self.action not in self.action_registry.actions: + raise ValidationError({'action': _("The log action '{}' has not been registered.").format(self.action)}) + + def __str__(self): + return "LogEntry %d: '%s' on '%s'" % ( + self.pk, self.action, self.object_verbose_name() + ) + + @cached_property + def user_display_name(self): + """ + Returns the display name of the associated user; + get_full_name if available and non-empty, otherwise get_username. + Defaults to 'system' when none is provided + """ + if self.user_id: + try: + user = self.user + except self._meta.get_field('user').related_model.DoesNotExist: + # User has been deleted + return _('user %(id)d (deleted)') % {'id': self.user_id} + + try: + full_name = user.get_full_name().strip() + except AttributeError: + full_name = '' + return full_name or user.get_username() + + else: + return _('system') + + @cached_property + def data(self): + """ + Provides deserialized data + """ + if self.data_json: + return json.loads(self.data_json) + else: + return {} + + @cached_property + def object_verbose_name(self): + model_class = self.content_type.model_class() + if model_class is None: + return self.content_type_id + + return model_class._meta.verbose_name.title + + def object_id(self): + raise NotImplementedError + + def format_message(self): + return self.action_registry.format_message(self) + + def format_comment(self): + return self.action_registry.format_comment(self) + + @property + def comment(self): + return self.format_comment()