Python
Send
import os
import requests
response = requests.post(
"https://api.epostix.com/v1/emails",
headers={
"Authorization": f"Bearer {os.environ['EPOSTIX_API_KEY']}",
"Idempotency-Key": f"order-{order.id}-confirmation",
},
json={
"subject": "It works",
"html": "<p>Sent from Python.</p>",
},
timeout=30,
)
if not response.ok:
error = response.json()
raise RuntimeError(f"{error['type']}: {error['message']} ({error['request_id']})")
print(response.json()["id"])
Always pass timeout. requests has none by default, so a stalled connection hangs the
worker until something else kills it.
Django
If you already have send_mail calls, repoint the backend and they keep working:
# settings.py
EMAIL_BACKEND = "django.core.mail.backends.smtp.EmailBackend"
EMAIL_HOST = "smtp.epostix.com"
EMAIL_PORT = 587
EMAIL_USE_TLS = True
EMAIL_HOST_USER = os.environ["EPOSTIX_SMTP_USER"]
EMAIL_HOST_PASSWORD = os.environ["EPOSTIX_SMTP_PASSWORD"]
Use the API for anything SMTP cannot express: tags, metadata, scheduling, idempotency, and reading the delivery timeline back.
Celery tasks need idempotency keys
A Celery task can run more than once. Acks-late, a worker restart, or a retry all redeliver it, and without a key each attempt sends another email.
@shared_task(bind=True, max_retries=3)
def send_receipt(self, order_id):
order = Order.objects.get(pk=order_id)
response = requests.post(
"https://api.epostix.com/v1/emails",
headers={
"Authorization": f"Bearer {settings.EPOSTIX_API_KEY}",
"Idempotency-Key": f"order-{order.id}-receipt",
},
json={...},
timeout=30,
)
if response.status_code >= 500:
raise self.retry(countdown=2 ** self.request.retries)
Derive the key from the order, never from uuid4() at call time. A fresh key on the retry
sends a second receipt, which is exactly what you were trying to avoid.
Next
- Test mode for local development and CI
- Idempotency