Celery¶

Инструкция для фоновой отправки почты через Celery (Python 3.10+). Письма уходят на SMTP Mailexam из воркера — удобно для регистрации, уведомлений и CI без блокировки HTTP-запроса.

Подходит вместе с Django, FastAPI или любым другим приложением: веб-слой ставит задачу в очередь, воркер отправляет письмо.

Что понадобится¶

Аккаунт Mailexam и проект с SMTP-учётными данными.
Брокер сообщений (в примере — Redis).
Python 3.10+ и виртуальное окружение.

Скопируйте из приветственного письма (или кабинета) для вашего проекта:

ВАШ_ЛОГИН — SMTP-логин (например, xxxxx);
ВАШ_ПАРОЛЬ — SMTP-пароль (уникальная пара к логину);
хост — ВАШ_ЛОГИН.mailexam.ru (совпадает с логином).

1. Зависимости¶

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

Запустите Redis локально (Docker):

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

2. Переменные окружения¶

Файл .env в корне проекта:

CELERY_BROKER_URL=redis://localhost:6379/0

MAILEXAM_LOGIN=ВАШ_ЛОГИН
MAILEXAM_PASSWORD=ВАШ_ПАРОЛЬ
MAILEXAM_PORT=587
MAIL_FROM=noreply@example.test

Хост SMTP: {MAILEXAM_LOGIN}.mailexam.ru.

3. Модуль отправки почты¶

Тот же smtplib, что в Flask — файл mail.py в корне проекта.

4. Приложение Celery и задача¶

# 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 из Celery",
) -> dict:
    send_test(to=to, subject=subject, body=body)
    return {"status": "ok", "to": to}

5. Запуск воркера и проверка¶

Терминал 1 — воркер:

celery -A celery_app worker --loglevel=info

Терминал 2 — постановка задачи:

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

Или из shell:

celery -A celery_app shell

from tasks import send_test_email
send_test_email.delay(
    to="user@example.test",
    subject="Проверка",
    body="Привет из очереди",
)

Письмо появится в кабинете Mailexam → ваш проект → входящие.

Интеграция с Django¶

Если почта уже настроена в Django, задача может вызывать 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(
        "Добро пожаловать",
        "Тест Mailexam через Celery + Django",
        None,
        [to],
    )

Запуск воркера в проекте Django:

celery -A config worker --loglevel=info

(config — модуль с Celery application, часто config/celery.py.)

6. Локальная разработка и CI¶

Среда	Рекомендация
`local`	Redis + `.env` с учётными данными Mailexam
CI	Сервис Redis в pipeline, секреты `MAILEXAM_LOGIN`, `MAILEXAM_PASSWORD`

Пример переменных в 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"

Для тестов без брокера и SMTP:

app.conf.task_always_eager = True

Задачи выполняются синхронно в том же процессе (удобно для unit-тестов, письмо всё равно уйдёт на Mailexam, если не подменить send_test).

После отправки из pipeline проверьте доставку через API Mailexam.

7. Типичные проблемы¶

Задача в статусе PENDING

Запущен ли воркер: celery -A celery_app worker --loglevel=info?
Совпадает ли CELERY_BROKER_URL у воркера и клиента, доступен ли Redis.

Ошибка SMTP в логах воркера

Хост {логин}.mailexam.ru, логин и пароль — пара из письма одного проекта.
Воркер наследует .env только если вы вызываете load_dotenv() в celery_app.py (как в примере).

Письмо не в кабинете

Смотрите входящие того же проекта Mailexam, чьи учётные данные в .env воркера.
Проверьте результат задачи: celery -A celery_app result <task_id>.