Source code for cms_qe_newsletter.models

from cms.models import CMSPlugin
from django.core.exceptions import ValidationError
from django.db import models
from django.utils import timezone
from django.utils.translation import ugettext_lazy as _


SUBSCRIBE = 1
UNSUBSCRIBE = 2

SERVICE_NONE = 1
SERVICE_MAILCHIMP = 2
SERVICES = (
    (SERVICE_NONE, _('None')),
    (SERVICE_MAILCHIMP, _('Mailchimp')),
)


[docs]class NewsletterPluginModel(CMSPlugin): """ Configuration model for newsletter plugin. Allows to set: * Title * Mail lists on mailchimp to which subscribers will be added * Show or hide full name * Require full name or not """ mailing_lists = models.ManyToManyField( 'MailingList', verbose_name=_('Maillist'), ) title = models.CharField( max_length=150, verbose_name=_('Title'), help_text=_('The title of subscribes that will be shown to users'), blank=True ) fullname_show = models.BooleanField( verbose_name=_('Show fullname fields'), default=False, ) fullname_require = models.BooleanField( verbose_name=_('Require fullname'), default=False, ) def __str__(self): return self.title
[docs] def clean(self): if not self.fullname_show and self.fullname_require: raise ValidationError(_('Can not hide and require a full name at the same time.'))
[docs]class MailingList(models.Model): """ Mailing list model which also synchronize with external services like Mailchimp. If ``external_id`` is not set then subscriber is not synchronized. """ name = models.CharField(max_length=50) external_service = models.PositiveSmallIntegerField(choices=SERVICES, default=SERVICE_NONE) external_id = models.CharField(max_length=10, null=True, blank=True) def __str__(self): return self.name
[docs] def clean(self): if self.external_service != SERVICE_NONE and not self.external_id: raise ValidationError(_('External ID is required when external service selected.'))
[docs]class Subscriber(models.Model): """ Subscriber model which also synchronize with external services like Mailchimp. If ``external_id`` is not set then subscriber is not synchronized. """ mailing_list = models.ForeignKey(MailingList, on_delete=models.CASCADE) email = models.EmailField() first_name = models.CharField(max_length=50, blank=True) last_name = models.CharField(max_length=50, blank=True) external_id = models.CharField(max_length=50, null=True, blank=True) def __str__(self): # pylint: disable=no-member return _('{} in list {}').format(self.email, self.mailing_list) # pylint: disable=signature-differs
[docs] def save(self, *args, **kwargs): self.subscribe() super().save(*args, **kwargs)
# pylint: disable=W0222 arguments-differ def delete(self, *args, **kwargs): self.unsubscribe() super().delete(*args, **kwargs)
[docs] def subscribe(self): """ Called before save to create task to sync action with external service. """ if self.mailing_list.external_service == SERVICE_NONE: return if not self.external_id: SubscribeTask.objects.create( mailing_list=self.mailing_list, email=self.email, first_name=self.first_name, last_name=self.last_name, type=SUBSCRIBE, )
[docs] def unsubscribe(self): """ Called before delete to create task to sync action with external service. """ if self.mailing_list.external_service == SERVICE_NONE: return if self.external_id: SubscribeTask.objects.create( mailing_list=self.mailing_list, email=self.email, first_name=self.first_name, last_name=self.last_name, external_id=self.external_id, type=UNSUBSCRIBE, ) else: # Not sync, just for case cancel any sync task. SubscribeTask.objects.filter( mailing_list=self.mailing_list, email=self.email, ).delete()
[docs]class SubscribeTask(models.Model): """ Item for task queue for subscribing and unsubscribing members on the external service like Mailchimp. The queue can is processed by ``cms-qe-newsletter-sync`` command. """ TASK_TYPES = ( (SUBSCRIBE, _('Subscribe')), (UNSUBSCRIBE, _('Unsubscribe')), ) # The subscriber's parameters are copied to the task, # in order not to lose them if the subscriber is removed from the database mailing_list = models.ForeignKey(MailingList, null=True, on_delete=models.CASCADE) email = models.EmailField() first_name = models.CharField(max_length=50, null=True) last_name = models.CharField(max_length=50, null=True) external_id = models.CharField(max_length=10, null=True) # Task params: type = models.PositiveSmallIntegerField(choices=TASK_TYPES) created = models.DateTimeField(auto_now=True) attempts = models.PositiveSmallIntegerField(default=0) last_error = models.TextField() def __str__(self): # pylint: disable=no-member return _('Task for email {}').format(self.email) def failure(self, error_message): self.attempts += 1 self.last_error = error_message self.save()
[docs] def should_process(self) -> bool: """ Checks whether task should be tried to process. With less attempts it tries more often, with more attempts it waits more time to not overwhelm resources. """ time_to_process = self.created + timezone.timedelta(seconds=self.attempts ** 4) return time_to_process <= timezone.now()