django-shared-utils/shared/utils/models/workflow.py

from django.conf import settings
from django.db import models
from django.utils import timezone
from django.utils.translation import gettext_lazy as _

#
# Workflow Helpers

"""
States:
- inactive (either active=False or publication date not reached)
- public (active or archived)
- active (public, not archived)
- archived (public, not active)
"""


class WorkflowQuerySet(models.QuerySet):
    @classmethod
    def exclude_inactive_filter(cls):
        """Usage: qs.exclude(**self.exclude_inactive_filter())"""
        return {
            "is_published": True,
            "publication_datetime__lte": timezone.now().date(),
        }

    @classmethod
    def active_filter(cls):
        return {
            "is_published": True,
            "publication_datetime__lte": timezone.now().date(),
            "archiving_datetime__gte": timezone.now().date(),
        }

    @classmethod
    def archived_filter(cls):
        return {
            "is_published": True,
            "publication_datetime__lte": timezone.now().date(),
            "archiving_datetime__lte": timezone.now().date(),
        }

    def public(self):
        # Active or archived
        return self.exclude(**self.exclude_inactive_filter())

    def active(self):
        return self.filter(**self.active_filter())

    def archived(self):
        return self.filter(**self.archived_filter())


class ManyToManyWorkflowQuerySet(WorkflowQuerySet):
    def _build_related_filters(self, filter_template):
        """
        Transforms all filter rules to match any related models which 
        itself is workflow managed.
        """
        filter = {}
        for field in self.model._meta.get_fields():
            if field.many_to_one and isinstance(field.related_model._meta.default_manager, WorkflowManager):
                field.name
                for path, value in filter_template.items():
                    filter[f'{field.name}__{path}'] = value
        return filter

    def public(self):
        # Active or archived
        return self.exclude(**self._build_related_filters(self.exclude_inactive_filter()))

    def active(self):
        return self.filter(**self._build_related_filters(self.active_filter()))

    def archived(self):
        return self.filter(**self._build_related_filters(self.archived_filter()))


class WorkflowManager(models.Manager.from_queryset(WorkflowQuerySet)):
    pass        


class ManyToManyWorkflowManager(models.Manager.from_queryset(ManyToManyWorkflowQuerySet)):
    pass


class WorkflowMixin(models.Model):
    creation_datetime = models.DateTimeField(auto_now_add=True)
    modification_datetime = models.DateTimeField(auto_now=True)
    creation_user = models.ForeignKey(settings.AUTH_USER_MODEL, null=True,
        editable=False,
        on_delete=models.SET_NULL, related_name='+')
    modification_user = models.ForeignKey(settings.AUTH_USER_MODEL, null=True,
        editable=False,
        on_delete=models.SET_NULL, related_name='+')

    is_published = models.BooleanField(_("Publicly visible"), default=False)
    publication_datetime = models.DateTimeField(_("Publication Date"), default=timezone.now)
    archiving_datetime = models.DateTimeField(_("Archiving Date"), default=timezone.datetime.max)

    objects = WorkflowManager()

    class Meta:
        abstract = True
        ordering = ['-publication_datetime']  # Most recent first

    @property
    def worflow_status(self):
        now = timezone.now()
        if not self.is_active or self.publication_datetime < now:
            return 'inactive'
        elif self.publication_datetime >= now and self.archiving_datetime < now:
            return 'active'
        else:
            return 'archived'


def display_is_active(obj):
    return obj.is_publised and obj.publication_datetime <= timezone.now()


display_is_active.boolean = True
display_is_active.short_description = _("Aktiv")