Skip to content

Celery

Guide for background mail delivery via Celery (Python 3.10+). Messages are sent to Mailexam SMTP from a worker — convenient for registration, notifications, and CI without blocking the HTTP request.

Works with Django, FastAPI, or any other application: the web layer enqueues a task, the worker sends the message.

What you need

  • A Mailexam account and a project with SMTP credentials.
  • A message broker (in this example — Redis).
  • Python 3.10+ and a virtual environment.

Copy from the welcome email (or dashboard) for your project:

  • YOUR_LOGIN — SMTP login (for example, xxxxx);
  • YOUR_PASSWORD — SMTP password (a unique pair with the login);
  • host — YOUR_LOGIN.mailexam.io (matches the login).

1. Dependencies

python3 -m venv .venv
source .venv/bin/activate
pip install celery redis python-dotenv

requirements.txt:

celery>=5.4
redis>=5.0
python-dotenv>=1.0

Run Redis locally (Docker):

docker run -d --name redis -p 6379:6379 redis:7-alpine

2. Environment variables

.env file in the project root:

CELERY_BROKER_URL=redis://localhost:6379/0

MAILEXAM_LOGIN=YOUR_LOGIN
MAILEXAM_PASSWORD=YOUR_PASSWORD
MAILEXAM_PORT=587
MAIL_FROM=noreply@example.test

SMTP host: {MAILEXAM_LOGIN}.mailexam.io.

3. Mail sending module

Same smtplib approach as in Flaskmail.py file in the project root.

4. Celery application and task

# celery_app.py
import os

from celery import Celery
from dotenv import load_dotenv

load_dotenv()

app = Celery(
    "mailexam",
    broker=os.environ.get("CELERY_BROKER_URL", "redis://localhost:6379/0"),
    include=["tasks"],
)
# tasks.py
from celery_app import app
from mail import send_test


@app.task(name="mail.send_test")
def send_test_email(
    to: str = "user@example.test",
    subject: str = "Celery + Mailexam",
    body: str = "Mailexam test from Celery",
) -> dict:
    send_test(to=to, subject=subject, body=body)
    return {"status": "ok", "to": to}

5. Running the worker and verification

Terminal 1 — worker:

celery -A celery_app worker --loglevel=info

Terminal 2 — enqueue a task:

python -c "from tasks import send_test_email; r = send_test_email.delay(); print(r.id)"

Or from shell:

celery -A celery_app shell
from tasks import send_test_email
send_test_email.delay(
    to="user@example.test",
    subject="Test",
    body="Hello from the queue",
)

The message will appear in the Mailexam dashboard → your project → inbox.

Django integration

If mail is already configured in Django, the task can call send_mail:

from celery import shared_task
from django.core.mail import send_mail


@shared_task
def send_welcome_email(to: str) -> None:
    send_mail(
        "Welcome",
        "Mailexam test via Celery + Django",
        None,
        [to],
    )

Run the worker in a Django project:

celery -A config worker --loglevel=info

(config — module with the Celery application, often config/celery.py.)

6. Local development and CI

Environment Recommendation
local Redis + .env with Mailexam credentials
CI Redis service in pipeline, secrets MAILEXAM_LOGIN, MAILEXAM_PASSWORD

Example variables in GitLab CI:

variables:
  CELERY_BROKER_URL: redis://redis:6379/0
  MAILEXAM_LOGIN: $MAILEXAM_LOGIN
  MAILEXAM_PASSWORD: $MAILEXAM_PASSWORD
  MAILEXAM_PORT: "587"
  MAIL_FROM: "noreply@example.test"

For tests without broker and SMTP:

app.conf.task_always_eager = True

Tasks run synchronously in the same process (convenient for unit tests; the message still goes to Mailexam unless you mock send_test).

After sending from the pipeline, verify delivery via the Mailexam API.

7. Common issues

Task stuck in PENDING status

  • Is the worker running: celery -A celery_app worker --loglevel=info?
  • Does CELERY_BROKER_URL match between worker and client, and is Redis reachable.

SMTP error in worker logs

  • Host {login}.mailexam.io, login and password — a pair from the email for one project.
  • The worker inherits .env only if you call load_dotenv() in celery_app.py (as in the example).

Message not in the dashboard

  • View the inbox of the same Mailexam project whose credentials are in the worker .env.
  • Check task result: celery -A celery_app result <task_id>.

See also