feat(brevo): envoyer les emails transactionnels via l'API Brevo v5 (#87)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
+42
-64
@@ -1,87 +1,65 @@
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from flask import Flask, render_template
|
||||
from flask_mail import Mail, Message
|
||||
from brevo import (
|
||||
Brevo,
|
||||
SendTransacEmailRequestSender,
|
||||
SendTransacEmailRequestToItem,
|
||||
)
|
||||
from brevo.core.api_error import ApiError
|
||||
|
||||
from src.utils import logger
|
||||
|
||||
TEMPLATES_DIR = Path(__file__).parent / "templates"
|
||||
|
||||
_mail: Mail | None = None
|
||||
_client: Brevo | None = None
|
||||
|
||||
|
||||
def init_mailer(app: Flask) -> None:
|
||||
global _mail
|
||||
app.config["MAIL_SERVER"] = os.getenv("SMTP_HOST", "")
|
||||
app.config["MAIL_PORT"] = int(os.getenv("SMTP_PORT", "587"))
|
||||
app.config["MAIL_USERNAME"] = os.getenv("SMTP_USERNAME") or None
|
||||
app.config["MAIL_PASSWORD"] = os.getenv("SMTP_PASSWORD") or None
|
||||
app.config["MAIL_USE_TLS"] = os.getenv("SMTP_USE_TLS", "True").lower() == "true"
|
||||
app.config["MAIL_DEFAULT_SENDER"] = os.getenv("MAIL_FROM", "noreply@decp.info")
|
||||
app.config["MAIL_SUPPRESS_SEND"] = (
|
||||
os.getenv("MAIL_SUPPRESS_SEND", os.getenv("DEVELOPMENT", "False")).lower()
|
||||
== "true"
|
||||
)
|
||||
# Inclure les templates du package auth dans la recherche Jinja
|
||||
if hasattr(app.jinja_loader, "searchpath"):
|
||||
app.jinja_loader.searchpath.append(str(TEMPLATES_DIR))
|
||||
else:
|
||||
import jinja2
|
||||
|
||||
app.jinja_loader = jinja2.ChoiceLoader(
|
||||
[app.jinja_loader, jinja2.FileSystemLoader(str(TEMPLATES_DIR))]
|
||||
)
|
||||
_mail = Mail(app)
|
||||
def init_mailer() -> None:
|
||||
"""Construit le client Brevo à partir des variables d'environnement."""
|
||||
global _client
|
||||
api_key = os.getenv("BREVO_API_KEY", "")
|
||||
sandbox = os.getenv("BREVO_SANDBOX", "").lower() == "true"
|
||||
headers = {"X-Sib-Sandbox": "drop"} if sandbox else None
|
||||
_client = Brevo(api_key=api_key, headers=headers)
|
||||
|
||||
|
||||
def _base_url() -> str:
|
||||
return os.getenv("APP_BASE_URL", "http://localhost:8050").rstrip("/")
|
||||
|
||||
|
||||
def _send(
|
||||
subject: str,
|
||||
recipient: str,
|
||||
txt_template: str,
|
||||
html_template: str,
|
||||
**ctx,
|
||||
) -> None:
|
||||
assert _mail is not None, "Mailer non initialisé (init_mailer(app) non appelé)"
|
||||
if _mail.suppress:
|
||||
suppress_cause = (
|
||||
"MAIL_SUPPRESS_SEND=true"
|
||||
if os.getenv("MAIL_SUPPRESS_SEND") is not None
|
||||
else "DEVELOPMENT=true"
|
||||
def _sender() -> SendTransacEmailRequestSender:
|
||||
return SendTransacEmailRequestSender(
|
||||
email=os.getenv("MAIL_FROM", "noreply@decp.info"),
|
||||
name=os.getenv("MAIL_FROM_NAME", "decp.info"),
|
||||
)
|
||||
|
||||
|
||||
def _template_id(env_var: str) -> int:
|
||||
raw = os.getenv(env_var, "")
|
||||
if not raw:
|
||||
raise RuntimeError(f"{env_var} non défini (template Brevo)")
|
||||
return int(raw)
|
||||
|
||||
|
||||
def _send_template(template_id: int, recipient: str, params: dict) -> None:
|
||||
assert _client is not None, "Mailer non initialisé (init_mailer() non appelé)"
|
||||
try:
|
||||
_client.transactional_emails.send_transac_email(
|
||||
template_id=template_id,
|
||||
params=params,
|
||||
sender=_sender(),
|
||||
to=[SendTransacEmailRequestToItem(email=recipient)],
|
||||
)
|
||||
logger.warning(
|
||||
"Email supprimé (%s) — non envoyé à %s : %s",
|
||||
suppress_cause,
|
||||
recipient,
|
||||
subject,
|
||||
except ApiError:
|
||||
logger.exception(
|
||||
"Échec d'envoi Brevo (template %s) à %s", template_id, recipient
|
||||
)
|
||||
msg = Message(subject=subject, recipients=[recipient])
|
||||
msg.body = render_template(txt_template, **ctx)
|
||||
msg.html = render_template(html_template, **ctx)
|
||||
_mail.send(msg)
|
||||
raise
|
||||
|
||||
|
||||
def send_verification_email(email: str, token: str) -> None:
|
||||
link = f"{_base_url()}/auth/verify-email?token={token}"
|
||||
_send(
|
||||
"Vérification de votre adresse email — decp.info",
|
||||
email,
|
||||
"emails/verify_email.txt",
|
||||
"emails/verify_email.html",
|
||||
link=link,
|
||||
)
|
||||
_send_template(_template_id("BREVO_TEMPLATE_VERIFY_ID"), email, {"link": link})
|
||||
|
||||
|
||||
def send_reset_email(email: str, token: str) -> None:
|
||||
link = f"{_base_url()}/reinitialiser-mot-de-passe?token={token}"
|
||||
_send(
|
||||
"Réinitialisation de votre mot de passe — decp.info",
|
||||
email,
|
||||
"emails/reset_password.txt",
|
||||
"emails/reset_password.html",
|
||||
link=link,
|
||||
)
|
||||
_send_template(_template_id("BREVO_TEMPLATE_RESET_ID"), email, {"link": link})
|
||||
|
||||
+4
-4
@@ -36,7 +36,7 @@ def init_auth(app: Flask) -> None:
|
||||
db.init_schema()
|
||||
db.purge_expired_tokens()
|
||||
|
||||
mailer.init_mailer(app)
|
||||
mailer.init_mailer()
|
||||
|
||||
_login_manager = LoginManager()
|
||||
_login_manager.login_view = "/connexion"
|
||||
@@ -49,8 +49,8 @@ def init_auth(app: Flask) -> None:
|
||||
|
||||
_csrf = CSRFProtect(app)
|
||||
|
||||
if not os.getenv("SMTP_HOST"):
|
||||
if not os.getenv("BREVO_API_KEY"):
|
||||
logger.warning(
|
||||
"SMTP_HOST non défini : les emails d'auth échoueront. "
|
||||
"Définissez les variables SMTP_* dans .env pour envoyer des emails."
|
||||
"BREVO_API_KEY non défini : les emails d'auth échoueront. "
|
||||
"Définissez les variables BREVO_* dans .env pour envoyer des emails."
|
||||
)
|
||||
|
||||
+25
-4
@@ -31,7 +31,28 @@ def client(app):
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mail_outbox(app):
|
||||
mail = app.extensions["mail"]
|
||||
with mail.record_messages() as outbox:
|
||||
yield outbox
|
||||
def mail_outbox(app, monkeypatch):
|
||||
from src.auth import mailer
|
||||
|
||||
calls = []
|
||||
|
||||
class _Msg:
|
||||
def __init__(self, call):
|
||||
self._call = call
|
||||
|
||||
@property
|
||||
def recipients(self):
|
||||
return [to.email for to in self._call["to"]]
|
||||
|
||||
class _FakeTransac:
|
||||
def send_transac_email(self, **kwargs):
|
||||
calls.append(_Msg(kwargs))
|
||||
|
||||
class _FakeClient:
|
||||
def __init__(self):
|
||||
self.transactional_emails = _FakeTransac()
|
||||
|
||||
monkeypatch.setattr(mailer, "_client", _FakeClient())
|
||||
monkeypatch.setenv("BREVO_TEMPLATE_VERIFY_ID", "11")
|
||||
monkeypatch.setenv("BREVO_TEMPLATE_RESET_ID", "22")
|
||||
yield calls
|
||||
|
||||
+52
-31
@@ -1,42 +1,63 @@
|
||||
import pytest
|
||||
from flask import Flask
|
||||
from flask_mail import Mail
|
||||
|
||||
from src.auth import mailer
|
||||
|
||||
|
||||
class _FakeTransac:
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
|
||||
def send_transac_email(self, **kwargs):
|
||||
self.calls.append(kwargs)
|
||||
|
||||
|
||||
class _FakeClient:
|
||||
def __init__(self):
|
||||
self.transactional_emails = _FakeTransac()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mail_app(monkeypatch, tmp_path):
|
||||
app = Flask(
|
||||
__name__,
|
||||
template_folder=str(
|
||||
(__import__("pathlib").Path(mailer.__file__).parent / "templates").resolve()
|
||||
),
|
||||
)
|
||||
app.config["MAIL_SUPPRESS_SEND"] = True
|
||||
app.config["MAIL_DEFAULT_SENDER"] = "noreply@decp.info"
|
||||
app.config["TESTING"] = True
|
||||
mail = Mail(app)
|
||||
monkeypatch.setattr(mailer, "_mail", mail)
|
||||
def fake_client(monkeypatch):
|
||||
client = _FakeClient()
|
||||
monkeypatch.setattr(mailer, "_client", client)
|
||||
monkeypatch.setenv("APP_BASE_URL", "http://localhost:8050")
|
||||
return app
|
||||
monkeypatch.setenv("BREVO_TEMPLATE_VERIFY_ID", "11")
|
||||
monkeypatch.setenv("BREVO_TEMPLATE_RESET_ID", "22")
|
||||
monkeypatch.setenv("MAIL_FROM", "noreply@decp.info")
|
||||
monkeypatch.setenv("MAIL_FROM_NAME", "decp.info")
|
||||
return client
|
||||
|
||||
|
||||
def test_send_verification_email(mail_app):
|
||||
with mail_app.app_context(), mail_app.test_request_context():
|
||||
with mail_app.extensions["mail"].record_messages() as outbox:
|
||||
mailer.send_verification_email("a@b.c", "TOKEN123")
|
||||
assert len(outbox) == 1
|
||||
msg = outbox[0]
|
||||
assert msg.recipients == ["a@b.c"]
|
||||
assert "/auth/verify-email?token=TOKEN123" in msg.body
|
||||
assert "/auth/verify-email?token=TOKEN123" in msg.html
|
||||
def test_send_verification_email(fake_client):
|
||||
mailer.send_verification_email("a@b.c", "TOKEN123")
|
||||
calls = fake_client.transactional_emails.calls
|
||||
assert len(calls) == 1
|
||||
call = calls[0]
|
||||
assert call["template_id"] == 11
|
||||
assert call["to"][0].email == "a@b.c"
|
||||
assert call["sender"].email == "noreply@decp.info"
|
||||
assert "/auth/verify-email?token=TOKEN123" in call["params"]["link"]
|
||||
|
||||
|
||||
def test_send_reset_email(mail_app):
|
||||
with mail_app.app_context(), mail_app.test_request_context():
|
||||
with mail_app.extensions["mail"].record_messages() as outbox:
|
||||
mailer.send_reset_email("a@b.c", "RESET456")
|
||||
assert len(outbox) == 1
|
||||
msg = outbox[0]
|
||||
assert "reinitialiser-mot-de-passe?token=RESET456" in msg.body
|
||||
def test_send_reset_email(fake_client):
|
||||
mailer.send_reset_email("a@b.c", "RESET456")
|
||||
calls = fake_client.transactional_emails.calls
|
||||
assert len(calls) == 1
|
||||
call = calls[0]
|
||||
assert call["template_id"] == 22
|
||||
assert "reinitialiser-mot-de-passe?token=RESET456" in call["params"]["link"]
|
||||
|
||||
|
||||
def test_init_mailer_builds_client(monkeypatch):
|
||||
monkeypatch.setenv("BREVO_API_KEY", "test-key")
|
||||
monkeypatch.delenv("BREVO_SANDBOX", raising=False)
|
||||
monkeypatch.setattr(mailer, "_client", None)
|
||||
mailer.init_mailer()
|
||||
assert mailer._client is not None
|
||||
|
||||
|
||||
def test_send_without_init_raises(monkeypatch):
|
||||
monkeypatch.setattr(mailer, "_client", None)
|
||||
monkeypatch.setenv("BREVO_TEMPLATE_VERIFY_ID", "11")
|
||||
with pytest.raises(AssertionError):
|
||||
mailer.send_verification_email("a@b.c", "TOKEN123")
|
||||
|
||||
Reference in New Issue
Block a user