activities.models のソースコード

# coding=utf-8
"""
"""

import pickle
from django.db import models
from django.db.models import Max
from django.core import serializers
from django.contrib.contenttypes.models import ContentType
from django.contrib.contenttypes.fields import GenericForeignKey
from django.utils.translation import ugettext as _


SNAPSHOT_CACHE_NAME = '_snapshot_cached'


class ActivityManager(models.Manager):

    def get_queryset(self):
        qs = super().get_queryset()
        qs = qs.prefetch_related('content_type')
        return qs

    def latests(self):
        """
        Return latest activities of each particular content_objects
        """
        # find created_at list of latest activities of each particular
        # content_objects
        # it use 'pk' instead of 'created_at' to filter latest while
        #   - more than two activities which has same 'created_at' is possible
        #   - newer activity have grater pk
        qs = super().get_queryset()
        qs = qs.values('content_type_id', 'object_id')
        qs = qs.annotate(pk=Max('pk'))
        pks = qs.values_list('pk', flat=True)
        # return activities corresponding to the latests
        return self.filter(pk__in=pks)

    def get_for_model(self, model):
        """
        Get activities related to the specified model
        """
        ct = ContentType.objects.get_for_model(model)
        return self.filter(content_type=ct)

    def get_for_object(self, obj):
        """
        Get activities related to the specified object
        """
        ct = ContentType.objects.get_for_model(obj)
        return self.filter(content_type=ct, object_id=obj.pk)


class Activity(models.Model):
    """
    A model wihch represent create/update/delete (and user specified status
    changes) activity of specified models
    """
    status = models.CharField(max_length=30)
    remarks = models.TextField(default='')

    content_type = models.ForeignKey(ContentType)
    object_id = models.PositiveIntegerField('Object ID')
    _snapshot = models.BinaryField(default=None, null=True)
    _content_object = GenericForeignKey()

    created_at = models.DateTimeField(auto_now_add=True)

    objects = ActivityManager()

    class Meta:
        ordering = ('-created_at',)
        verbose_name = _('Activity')
        verbose_name_plural = _('Activities')

    def __repr__(self):
        return "<Activity: {}:{}:{}>".format(self.content_type.model,
                                             self.object_id,
                                             self.status)

    @property
    def mediator(self):
        from .registry import registry
        return registry.get(self)

    @property
    def snapshot(self):
        """
        Get a pickled object from database
        """
        if not hasattr(self, SNAPSHOT_CACHE_NAME):
            snapshot = None
            if self._snapshot:
                serialized_obj = pickle.loads(self._snapshot)
                if serialized_obj and isinstance(serialized_obj, dict):
                    snapshot = self.mediator.deserialize_snapshot(
                        serialized_obj
                    )
                else:
                    snapshot = serialized_obj
            setattr(self, SNAPSHOT_CACHE_NAME, snapshot)
        return getattr(self, SNAPSHOT_CACHE_NAME)

    @snapshot.setter
    def snapshot(self, value):
        """
        Set object to database as pickled object
        """
        if value:
            serialized_obj = self.mediator.serialize_snapshot(value)
            snapshot = pickle.dumps(serialized_obj)
        else:
            snapshot = None
        self._snapshot = snapshot
        # delete cached object
        if hasattr(self, SNAPSHOT_CACHE_NAME):
            delattr(self, SNAPSHOT_CACHE_NAME)

    @property
    def previous(self):
        """
        Get a previous activity instance which have same content_type and
        object_id as this activity instance.
        This is a shortcut property of the following code

            qs = activity.get_previous_activities()
            previous = qs.first()

        """
        qs = self.get_previous_activities()
        return qs.first()

    def get_related_activities(self):
        """
        Get a queryset of activities which have same content_type and object_id
        as this activity instance except the activity itself.
        Note that the queryset is cached in the instance thus you may need to
        re-get the instance from a database to update the cache.
        """
        cache_name = '_related_cache'
        if not hasattr(self, cache_name):
            qs = Activity.objects.filter(content_type=self.content_type,
                                         object_id=self.object_id)
            if self.pk:
                qs = qs.exclude(pk=self.pk)
            setattr(self, cache_name, qs)
        return getattr(self, cache_name)

    def get_previous_activities(self):
        """
        Get a queryset of activities which is smilar to the queryset returned
        by `get_related_activities` method but only older activities are
        contained.
        """
        qs = self.get_related_activities()
        if self.pk:
            qs = qs.exclude(pk__gte=self.pk)
        return qs

    def get_next_activities(self):
        """
        Get a queryset of activities which is smilar to the queryset returned
        by `get_related_activities` method but only newer activities are
        contained.
        """
        qs = self.get_related_activities()
        if self.pk:
            qs = qs.exclude(pk__lte=self.pk)
        return qs