"""Email service — SMTP delivery with HTML templates and delivery logging."""
import smtplib
import logging
import time
import uuid
from datetime import datetime, timezone
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from pathlib import Path
from typing import Optional
from app.config import settings
logger = logging.getLogger(__name__)
TEMPLATE_DIR = Path(__file__).parent.parent / "templates"
def _render_template(template_name: str, context: dict) -> str:
"""Render a Jinja2 HTML email template. Returns empty string on failure."""
try:
from jinja2 import Environment, FileSystemLoader
env = Environment(loader=FileSystemLoader(str(TEMPLATE_DIR)), autoescape=True)
tmpl = env.get_template(template_name)
return tmpl.render(**context)
except Exception as e:
logger.warning("Template render failed (%s): %s", template_name, e)
return ""
def _log_email(
to: str,
subject: str,
email_type: str,
status: str,
school_id: Optional[str] = None,
error_message: Optional[str] = None,
) -> None:
"""Write an email delivery record to email_logs. Best-effort — never raises."""
try:
import os
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.models.email_log import EmailLog, EmailStatus, EmailType
db_url = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@db:5432/taptrack_hub")
engine = create_engine(
db_url.replace("postgresql+asyncpg://", "postgresql://"),
pool_pre_ping=True,
pool_size=1,
max_overflow=0,
)
Session = sessionmaker(bind=engine)
db = Session()
try:
# Map type string to enum, default to 'other'
try:
etype = EmailType(email_type)
except ValueError:
etype = EmailType.other
log = EmailLog(
id=str(uuid.uuid4()),
school_id=school_id,
to_email=to,
subject=subject,
email_type=etype,
status=EmailStatus(status),
error_message=error_message,
sent_at=datetime.now(timezone.utc) if status == "sent" else None,
created_at=datetime.now(timezone.utc),
)
db.add(log)
db.commit()
finally:
db.close()
engine.dispose()
except Exception as e:
logger.debug("Email log write failed: %s", e)
def send_email(
to: str,
subject: str,
body: str,
html: Optional[str] = None,
template_name: Optional[str] = None,
context: Optional[dict] = None,
email_type: str = "other",
school_id: Optional[str] = None,
max_retries: int = 3,
) -> bool:
"""
Send an email via configured SMTP.
Priority order for HTML content:
1. `html` parameter (raw HTML string)
2. `template_name` + `context` (rendered Jinja2 template)
3. `body` only (plain text)
Logs every attempt to email_logs table.
Retries up to `max_retries` times on transient SMTP failure.
Returns True on success.
"""
if not settings.SMTP_HOST:
logger.warning("SMTP not configured — would send to %s: %s", to, subject)
_log_email(to, subject, email_type, "failed", school_id, "SMTP not configured")
return False
# Resolve HTML content
html_content = html
if not html_content and template_name:
ctx = context or {}
ctx.setdefault("subject", subject)
ctx.setdefault("portal_url", settings.HUB_BASE_URL)
html_content = _render_template(template_name, ctx)
last_error: Optional[str] = None
for attempt in range(1, max_retries + 1):
try:
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = settings.SMTP_FROM
msg["To"] = to
msg.attach(MIMEText(body, "plain"))
if html_content:
msg.attach(MIMEText(html_content, "html"))
with smtplib.SMTP(settings.SMTP_HOST, settings.SMTP_PORT) as smtp:
smtp.ehlo()
smtp.starttls()
smtp.ehlo()
if settings.SMTP_USER:
smtp.login(settings.SMTP_USER, settings.SMTP_PASSWORD)
smtp.sendmail(settings.SMTP_FROM, [to], msg.as_string())
logger.info("Email sent to %s: %s", to, subject)
_log_email(to, subject, email_type, "sent", school_id)
return True
except smtplib.SMTPException as e:
last_error = str(e)
logger.warning("SMTP attempt %d/%d failed for %s: %s", attempt, max_retries, to, e)
if attempt < max_retries:
time.sleep(2 ** attempt) # exponential backoff: 2s, 4s
except Exception as e:
last_error = str(e)
logger.error("Email send failed for %s: %s", to, e)
break # Non-SMTP errors don't retry
_log_email(to, subject, email_type, "failed", school_id, last_error)
return False