Why Gemfury? Push, build, and install  RubyGems npm packages Python packages Maven artifacts PHP packages Go Modules Bower components Debian packages RPM packages NuGet packages

nickfrez / django-emails   python

Repository URL to install this package:

Version: 0.0.5 

/ django_emails / models.py

from datetime import timedelta

from django.conf import settings
from django.db import DataError
from django.db import models
from django.db.models import Case
from django.db.models import F
from django.db.models import Value
from django.db.models import When
from django.utils import crypto
from django.utils import timezone

from .fields import NullTrueField


class EmailQuerySet(models.QuerySet):
  def user(self, user):
    return self.filter(user=user)

  def address(self, address):
    return self.filter(address__iexact=address)

  def active(self):
    return self.filter(active=True)

  def primary(self):
    return self.filter(primary=True)

  def confirmed(self):
    return self.filter(confirmed_at__isnull=False)


class Email(models.Model):
  # TODO(nick): We might not want to tie this too closely to the user model.
  #   It's feasible that a project may want emails to belong to another model,
  #   such as an organization, group, etc.
  user = models.ForeignKey(settings.AUTH_USER_MODEL, null=True,
                           related_name='emails')
  address = models.EmailField()
  primary = NullTrueField(default=None)
  active = NullTrueField(default=True)
  confirmation_sent_at = models.DateTimeField(auto_now_add=True)
  confirmed_at = models.DateTimeField(null=True, blank=True)
  archived_at = models.DateTimeField(null=True, blank=True)

  class Meta:
    abstract = True
    unique_together = (
      ('user', 'primary'),
      ('address', 'active'),
    )
    verbose_name = 'email'
    verbose_name_plural = 'emails'

  objects = EmailQuerySet.as_manager()

  MAX_CONFIRMATION_AGE = timedelta(days=7)

  def save(self, *args, **kwargs):
    if not self.address:
      raise DataError('address cannot be blank.')
    super(Email, self).save(*args, **kwargs)

  def __str__(self):
    return unicode(self.address)

  # TODO(nick): Right now this assumes that when marking an email as primary,
  #   we want to unset any existing primary emails.  That's not necessarily
  #   going to be true from project-to-project.  This behavior is something
  #   that should be left to the project to implement.
  # def make_primary(self, refresh=True):
  #   """Make this email the primary email.

  #   If a primary email already exists, it is set to non-primary.

  #   Args:

  #     refresh(bool):  Refresh the instance from the db after update?
  #   """
  #   Email.objects.user(self.user).update(
  #     primary=Case(
  #       When(id=self.id,
  #            then=Value(True)),
  #       default=Value(None),
  #     ))
  #   if refresh:
  #     self.refresh_from_db()

  def archive(self, refresh=True):
    """Remove this email, but only if it is not the primary.

    Args:

      refresh(bool):  Refresh the instance from the db after update?
    """
    Email.objects.filter(pk=self.pk).update(
      archived_at=Case(
        When(primary__isnull=True,
             then=Value(timezone.now())),
        default=F('archived_at'),
      ),
      active=Case(
        When(primary__isnull=True,
             then=Value(None)),
        default=F('active'),
      ),
    )
    if refresh:
      self.refresh_from_db()

  def confirmation_code(self):
    """Return the current confirmation code for this email address."""
    return crypto.salted_hmac(str(self.confirmation_sent_at),
                              self.address).hexdigest()

  def new_confirmation_code(self):
    """Update and return a new confirmation code.

    Note that we don't ever store the confirmation code, we simply generate
    it as a hash of the address and the confirmation_sent_at timestamp.

    This means that "creating a new code" is a simple matter of updating the
    confirmation_sent_at field, and that all previous codes will therefore
    be invalidated.
    """
    self.confirmation_sent_at = timezone.now()
    self.save(update_fields=['confirmation_sent_at'])
    return self.confirmation_code()

  def validate_confirmation_code(self, code):
    timeout = self.confirmation_sent_at + self.MAX_CONFIRMATION_AGE
    if timezone.now() > timeout:
      return False
    return crypto.constant_time_compare(code, self.confirmation_code())