from datetime import datetime from django.conf import settings from django.db import models from django.dispatch import receiver from django.utils.translation import gettext_lazy as _ from model_utils import Choices from polymorphic.base import PolymorphicModelBase from polymorphic.managers import PolymorphicManager from polymorphic.models import PolymorphicModel from polymorphic.query import PolymorphicQuerySet from rules.contrib.models import RulesModelBaseMixin, RulesModelMixin from outbo.apps.core.fields import ( RemovalDateTimeField, CreationDateTimeField, ModificationDateTimeField, ) from outbo.apps.core.utils import get_now EmailLogStatus = Choices("sending", "sent", "queued", "invalid", "rejected", "failed", "unknown") class EmailLog(models.Model): sender = models.CharField(max_length=254, blank=False, null=False) receiver = models.CharField(max_length=254, blank=False, null=False) time = models.DateTimeField(blank=False, null=False) status = models.CharField(max_length=20, blank=False, null=False, choices=EmailLogStatus) if "anymail" in settings.INSTALLED_APPS: from anymail.signals import pre_send, post_send from anymail.message import AnymailStatus from django.core.mail import EmailMessage def log_email_message(sender: str, receiver: str, time: datetime, status: str): try: EmailLog.objects.create(sender=sender, receiver=receiver, time=time, status=status) except: pass @receiver(pre_send) def handle_pre_send(sender, message: EmailMessage, **kwargs): log_email_message( sender=str(message.from_email), receiver=", ".join(message.to), time=get_now(), status=EmailLogStatus.sending, ) @receiver(post_send) def handle_post_send(sender, message: EmailMessage, status: AnymailStatus, esp_name, **kwargs): for email, recipient_status in status.recipients.items(): log_email_message( sender=str(message.from_email), receiver=email, time=get_now(), status=recipient_status.status or "failed", ) class SoftDeletableQuerySet(models.QuerySet): def delete(self): return super().update(deleted_at=get_now()) def hard_delete(self): return super().delete() class SoftDeletableManager(models.Manager): queryset_class = SoftDeletableQuerySet def __init__(self, *args, **kwargs): self.alive_only = kwargs.pop("alive_only", True) super().__init__(*args, **kwargs) def get_queryset(self): if self.alive_only: return self.queryset_class(self.model).filter(deleted_at=None) return self.queryset_class(self.model) class TimestampableDeletableMixin(models.Model): objects = SoftDeletableManager() all_objects = SoftDeletableManager(alive_only=False) created_at = CreationDateTimeField(_("Created at")) updated_at = ModificationDateTimeField(_("Updated at")) deleted_at = RemovalDateTimeField(_("Deleted at")) class Meta: abstract = True def delete(self, using=None, **kwargs): self.deleted_at = get_now() # TODO: handle autodeletion of nested stuff self.save(using=using) def __str__(self): return super().__str__() if self.deleted_at is None else f"{super().__str__()} (deleted)" class RulesPolymorphicModelBase(RulesModelBaseMixin, PolymorphicModelBase): pass class RulesPolymorphicModel(RulesModelMixin, PolymorphicModel, metaclass=RulesPolymorphicModelBase): class Meta: abstract = True class SoftDeletablePolymorphicQuerySet(SoftDeletableQuerySet, PolymorphicQuerySet): pass class SoftDeletablePolymorphicManager(SoftDeletableManager, PolymorphicManager): queryset_class = SoftDeletablePolymorphicQuerySet