Public Access
adminandClaude Sonnet 4.6
5a41402644
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
103 lines
2.7 KiB
Python
103 lines
2.7 KiB
Python
"""
|
|
Email backends.
|
|
|
|
R2-B spike: only ConsoleEmailBackend is functional. SendGrid is a stub
|
|
intentionally left to be wired in R3-C.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional, Protocol
|
|
|
|
from sendgrid import SendGridAPIClient
|
|
from sendgrid.helpers.mail import Mail, ReplyTo
|
|
|
|
from app.config import settings
|
|
|
|
|
|
# Repository convention: backend/var/email_outbox.jsonl
|
|
# This module lives at backend/app/services/email.py — go up two levels to
|
|
# reach backend/, then var/.
|
|
_BACKEND_ROOT = Path(__file__).resolve().parent.parent.parent
|
|
OUTBOX_PATH = _BACKEND_ROOT / "var" / "email_outbox.jsonl"
|
|
|
|
|
|
class EmailBackend(Protocol):
|
|
def send(
|
|
self,
|
|
to: str,
|
|
subject: str,
|
|
html: str,
|
|
text: Optional[str] = None,
|
|
) -> None:
|
|
...
|
|
|
|
|
|
class ConsoleEmailBackend:
|
|
"""Dev/spike backend. Prints to stdout AND appends a JSON line to
|
|
backend/var/email_outbox.jsonl so the round-trip can be inspected
|
|
after the fact.
|
|
"""
|
|
|
|
def send(
|
|
self,
|
|
to: str,
|
|
subject: str,
|
|
html: str,
|
|
text: Optional[str] = None,
|
|
) -> None:
|
|
record = {
|
|
"ts": datetime.now(timezone.utc).isoformat(),
|
|
"to": to,
|
|
"subject": subject,
|
|
"html": html,
|
|
"text": text,
|
|
}
|
|
print(
|
|
f"[ConsoleEmailBackend] -> {to} | {subject}",
|
|
file=sys.stdout,
|
|
flush=True,
|
|
)
|
|
OUTBOX_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
with OUTBOX_PATH.open("a", encoding="utf-8") as f:
|
|
f.write(json.dumps(record) + "\n")
|
|
|
|
|
|
class SendGridEmailBackend:
|
|
def __init__(self) -> None:
|
|
self._client = SendGridAPIClient(settings.SENDGRID_API_KEY)
|
|
self._from_email = settings.SENDGRID_FROM_EMAIL
|
|
self._reply_to = settings.SENDGRID_REPLY_TO
|
|
|
|
def send(
|
|
self,
|
|
to: str,
|
|
subject: str,
|
|
html: str,
|
|
text: Optional[str] = None,
|
|
) -> None:
|
|
message = Mail(
|
|
from_email=self._from_email,
|
|
to_emails=to,
|
|
subject=subject,
|
|
html_content=html,
|
|
plain_text_content=text,
|
|
)
|
|
message.reply_to = ReplyTo(self._reply_to)
|
|
response = self._client.send(message)
|
|
if response.status_code >= 400:
|
|
raise RuntimeError(
|
|
f"SendGrid error {response.status_code}: {response.body}"
|
|
)
|
|
|
|
|
|
def get_email_backend() -> EmailBackend:
|
|
backend = (settings.EMAIL_BACKEND or "console").lower()
|
|
if backend == "sendgrid":
|
|
return SendGridEmailBackend()
|
|
return ConsoleEmailBackend()
|