netbox/netbox/core/models/change_logging.py

from functools import cached_property

from django.conf import settings
from django.contrib.contenttypes.fields import GenericForeignKey
from django.core.exceptions import ValidationError
from django.db import models
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from mptt.models import MPTTModel

from core.choices import ObjectChangeActionChoices
from core.querysets import ObjectChangeQuerySet
from netbox.models.features import ChangeLoggingMixin, has_feature
from utilities.data import shallow_compare_dict

__all__ = (
    'ObjectChange',
)


class ObjectChange(models.Model):
    """
    Record a change to an object and the user account associated with that change. A change record may optionally
    indicate an object related to the one being changed. For example, a change to an interface may also indicate the
    parent device. This will ensure changes made to component models appear in the parent model's changelog.
    """
    time = models.DateTimeField(
        verbose_name=_('time'),
        auto_now_add=True,
        editable=False,
        db_index=True
    )
    user = models.ForeignKey(
        to=settings.AUTH_USER_MODEL,
        on_delete=models.SET_NULL,
        related_name='changes',
        blank=True,
        null=True
    )
    user_name = models.CharField(
        verbose_name=_('user name'),
        max_length=150,
        editable=False
    )
    request_id = models.UUIDField(
        verbose_name=_('request ID'),
        editable=False,
        db_index=True
    )
    action = models.CharField(
        verbose_name=_('action'),
        max_length=50,
        choices=ObjectChangeActionChoices
    )
    changed_object_type = models.ForeignKey(
        to='contenttypes.ContentType',
        on_delete=models.PROTECT,
        related_name='+'
    )
    changed_object_id = models.PositiveBigIntegerField()
    changed_object = GenericForeignKey(
        ct_field='changed_object_type',
        fk_field='changed_object_id'
    )
    related_object_type = models.ForeignKey(
        to='contenttypes.ContentType',
        on_delete=models.PROTECT,
        related_name='+',
        blank=True,
        null=True
    )
    related_object_id = models.PositiveBigIntegerField(
        blank=True,
        null=True
    )
    related_object = GenericForeignKey(
        ct_field='related_object_type',
        fk_field='related_object_id'
    )
    object_repr = models.CharField(
        max_length=200,
        editable=False
    )
    message = models.CharField(
        verbose_name=_('message'),
        max_length=200,
        editable=False,
        blank=True
    )
    prechange_data = models.JSONField(
        verbose_name=_('pre-change data'),
        editable=False,
        blank=True,
        null=True
    )
    postchange_data = models.JSONField(
        verbose_name=_('post-change data'),
        editable=False,
        blank=True,
        null=True
    )

    objects = ObjectChangeQuerySet.as_manager()

    class Meta:
        ordering = ['-time']
        indexes = (
            models.Index(fields=('changed_object_type', 'changed_object_id')),
            models.Index(fields=('related_object_type', 'related_object_id')),
        )
        verbose_name = _('object change')
        verbose_name_plural = _('object changes')

    def __str__(self):
        return '{} {} {} by {}'.format(
            self.changed_object_type,
            self.object_repr,
            self.get_action_display().lower(),
            self.user_name
        )

    def clean(self):
        super().clean()

        # Validate the assigned object type
        if not has_feature(self.changed_object_type, 'change_logging'):
            raise ValidationError(
                _("Change logging is not supported for this object type ({type}).").format(
                    type=self.changed_object_type
                )
            )

    def save(self, *args, **kwargs):

        # Record the user's name and the object's representation as static strings
        if not self.user_name:
            self.user_name = self.user.username
        if not self.object_repr:
            self.object_repr = str(self.changed_object)

        return super().save(*args, **kwargs)

    def get_absolute_url(self):
        return reverse('core:objectchange', args=[self.pk])

    def get_action_color(self):
        return ObjectChangeActionChoices.colors.get(self.action)

    @cached_property
    def has_changes(self):
        return self.prechange_data != self.postchange_data

    @cached_property
    def diff_exclude_fields(self):
        """
        Return a set of attributes which should be ignored when calculating a diff
        between the pre- and post-change data. (For instance, it would not make
        sense to compare the "last updated" times as these are expected to differ.)
        """
        model = self.changed_object_type.model_class()
        attrs = set()

        # Exclude auto-populated change tracking fields
        if issubclass(model, ChangeLoggingMixin):
            attrs.update({'created', 'last_updated'})

        # Exclude MPTT-internal fields
        if issubclass(model, MPTTModel):
            attrs.update({'level', 'lft', 'rght', 'tree_id'})

        return attrs

    def get_clean_data(self, prefix):
        """
        Return only the pre-/post-change attributes which are relevant for calculating a diff.
        """
        ret = {}
        change_data = getattr(self, f'{prefix}_data') or {}
        for k, v in change_data.items():
            if k not in self.diff_exclude_fields and not k.startswith('_'):
                ret[k] = v
        return ret

    @cached_property
    def prechange_data_clean(self):
        return self.get_clean_data('prechange')

    @cached_property
    def postchange_data_clean(self):
        return self.get_clean_data('postchange')

    def diff(self):
        """
        Return a dictionary of pre- and post-change values for attributes which have changed.
        """
        prechange_data = self.prechange_data_clean
        postchange_data = self.postchange_data_clean

        # Determine which attributes have changed
        if self.action == ObjectChangeActionChoices.ACTION_CREATE:
            changed_attrs = sorted(postchange_data.keys())
        elif self.action == ObjectChangeActionChoices.ACTION_DELETE:
            changed_attrs = sorted(prechange_data.keys())
        else:
            # TODO: Support deep (recursive) comparison
            changed_data = shallow_compare_dict(prechange_data, postchange_data)
            changed_attrs = sorted(changed_data.keys())

        return {
            'pre': {
                k: prechange_data.get(k) for k in changed_attrs
            },
            'post': {
                k: postchange_data.get(k) for k in changed_attrs
            },
        }