Django integration: send transactional email

Django's send_mail goes through EMAIL_BACKEND, which is SMTP by default. Inboxili is API-only, so you call it from a service function instead.

Settings

settings.py:

import os

INBOXILI_API_KEY = os.environ["INBOXILI_API_KEY"]
INBOXILI_BASE_URL = os.environ.get("INBOXILI_BASE_URL", "https://api.inboxili.com/api/v1")
INBOXILI_FROM_EMAIL = os.environ.get("INBOXILI_FROM_EMAIL", "hello@yourdomain.com")
INBOXILI_FROM_NAME = "Acme"

The service

mail/inboxili.py:

import requests
from django.conf import settings


class InboxiliError(Exception):
    def __init__(self, status, code, message):
        super().__init__(message)
        self.status, self.code = status, code


def send(*, to, subject=None, html_body=None, text_body=None, template_id=None, template_data=None,
         from_email=None, from_name=None):
    payload = {
        "to": to,
        "from_email": from_email or settings.INBOXILI_FROM_EMAIL,
        "from_name": from_name or settings.INBOXILI_FROM_NAME,
        "subject": subject,
        "html_body": html_body,
        "text_body": text_body,
        "template_id": template_id,
        "template_data": template_data or {},
    }
    payload = {k: v for k, v in payload.items() if v is not None}

    resp = requests.post(
        f"{settings.INBOXILI_BASE_URL}/transactional/send",
        json=payload,
        headers={"Authorization": f"Bearer {settings.INBOXILI_API_KEY}"},
        timeout=10,
    )
    if not resp.ok:
        err = (resp.json() if resp.content else {}).get("error", {})
        raise InboxiliError(resp.status_code, err.get("code", "http_error"), err.get("message", resp.reason))
    return resp.json()

Send from a Celery task

from celery import shared_task
from django.contrib.auth import get_user_model
from mail import inboxili


@shared_task(bind=True, max_retries=4)
def send_welcome_email(self, user_id):
    user = get_user_model().objects.get(pk=user_id)
    try:
        inboxili.send(
            to=user.email,
            subject="Welcome, {{first_name}}",
            html_body="<p>Hi {{first_name}}, your account is ready.</p>",
            template_data={"first_name": user.first_name or "there"},
        )
    except inboxili.InboxiliError as e:
        if e.status == 429:
            raise self.retry(exc=e, countdown=2 ** self.request.retries * 5)
        raise

Trigger it after the transaction commits so the task never runs against a user row that was rolled back:

from django.db import transaction

transaction.on_commit(lambda: send_welcome_email.delay(user.pk))

Error handling

| status | code | Do | |---|---|---| | 401 | unauthorized | Check the key. | | 403 | forbidden | Check scope and IP rules. | | 422 | validation_error, sender_not_verified, send_failed | Fix the request. | | 429 | rate_limited | Retry with backoff (done above). |

requests.Timeout means the outcome is unknown. With no idempotency key, retrying it can send a second email.

Testing

from unittest.mock import patch, MagicMock
from django.test import TestCase, override_settings
from mail import inboxili


@override_settings(INBOXILI_API_KEY="ik_live_test")
class InboxiliTests(TestCase):
    @patch("mail.inboxili.requests.post")
    def test_send(self, post):
        post.return_value = MagicMock(ok=True, json=lambda: {"status": "sent", "message_id": "m1"})
        out = inboxili.send(to="a@b.co", subject="s", html_body="<p>x</p>")
        self.assertEqual(out["message_id"], "m1")
        self.assertEqual(post.call_args.kwargs["json"]["to"], "a@b.co")

Production notes

• Keep the key in the environment, not in settings.py.
• Keep Django's EMAIL_BACKEND for admin error mails, if you use them, but be aware those do not go through Inboxili.
• Webhooks: use a csrf-exempt view and compute the hex HMAC-SHA256 of request.body with your webhook secret and compare it to the X-Inboxili-Signature header using hmac.compare_digest.