Django: How to send email without Celery
Lightweight emails without Celery. And without a request-response cycle.
Published: May 28, 2022 Sponsored App StoreI have to admit that I found sending emails in Django more difficult than I expected since other more complex stuff seemed easier.
In this post, I want to share a lightweight approach that I am successfully using for a somewhat lower email volume.
While the most basic solution is just sending email in your view code (before you return a response), this is problematic. It will make the response much slower, and if the email fails to send, you have no real means of sending it again.
Management command + CRON to the rescue
I am currently quite happy with basic management commands combined with CRON schedule that runs every few hours and sends emails.
We can keep track of sent emails with records in the database. For example, when I send an email notification that a trial expires soon, I create a record in the database to ensure the email is not sent twice.
The model looks something like this:
class SentEmailRecord(models.Model):
timestamp = models.DateTimeField(auto_now_add=True)
email_type = models.CharField(blank=False, max_length=50)
to_user = models.ForeignKey(to=CustomUser, blank=True, null=True, on_delete=models.SET_NULL)
The email_type
is just a field to save the type I have as a constant in the project. For example, trial-expiring
is one type.
I am using this approach because I am sending the trial notification only to people who interacted with my project. And this solution works pretty well.
The other piece of the puzzle is the management command:
class Command(BaseCommand):
help = 'Sends info emails about expiring trial'
def handle(self, *args, **options):
# omitted query to get relevant users
for user in users_to_notify:
email_sent = SentEmailRecord.objects.filter(to_user=user, email_type=SentEmailRecord.TRIAL_EXPIRING_TYPE).exists()
if email_sent:
continue
sent_successfully = send_trial_expiring_email(user)
if sent_successfully:
SentEmailRecord.objects.get_or_create(to_user=user, email_type=SentEmailRecord.TRIAL_EXPIRING_TYPE)
If the sending succeeds, a new SentEmailRecord
is created to prevent another email from going off. Since this command runs a few times each day, there is a guaranteed retry for emails that might have failed for some reason.
Pending email approach
Another management command approach with CRON scheduling is to “reverse” the model logic I described above. You would first save a record of email you want to send, then send it and mark it as sent.
This would work well for cases where you are sure that the system will send the email. For example, a registration confirmation.
Instead of sending the email directly from the view code, you would create a new pending email, which would get sent later.
Depending on your needs, you can store the email type without the content if the email is not going to vary.
class PendingEmail(models.Model):
was_sent = models.BooleanField(default=False)
email_type = models.CharField(blank=False, max_length=50)
to_user = models.ForeignKey(to=CustomUser, blank=True, null=True, on_delete=models.SET_NULL)
As you can see, it looks almost the same. Now in your register view, you could do something like:
PendingEmail.objects.create(to_user=new_user, email_type="register-confirm")
And then have a management script that finds PendingEmail
objects that haven’t been sent yet, sends them, and marks the was_sent
property.
class Command(BaseCommand):
help = 'Handles pending email'
def handle(self, *args, **options):
pending_emails = PendingEmail.objects.filter(was_sent=False)
for email in pending_emails:
sent_successfully = send_pending_email(email)
if sent_successfully:
email.was_sent = True
email.save()
And you have a basic system for emails without going with a complicated task queue. Don’t forget that the models shown here are mostly meant as an example. Customize them to your needs.