feat: Add SMTP support for self-hosted Postal mail server
Some checks are pending
Deploy Elmeg / deploy (push) Waiting to run

This commit is contained in:
fullsizemalt 2025-12-23 17:19:22 -08:00
parent 0af64f5862
commit 9c92eb7953

View file

@ -1,9 +1,12 @@
""" """
Email Service - Mailgun (primary) with AWS SES fallback Email Service - Postal SMTP (primary), Mailgun, or AWS SES fallback
""" """
import os import os
import httpx import httpx
import secrets import secrets
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import Optional from typing import Optional
@ -18,7 +21,14 @@ except ImportError:
class EmailService: class EmailService:
def __init__(self): def __init__(self):
# Mailgun settings (primary) # Postal SMTP settings (primary - self-hosted)
self.smtp_host = os.getenv("SMTP_HOST")
self.smtp_port = int(os.getenv("SMTP_PORT", "25"))
self.smtp_username = os.getenv("SMTP_USERNAME")
self.smtp_password = os.getenv("SMTP_PASSWORD")
self.smtp_use_tls = os.getenv("SMTP_USE_TLS", "true").lower() == "true"
# Mailgun settings (alternative)
self.mailgun_api_key = os.getenv("MAILGUN_API_KEY") self.mailgun_api_key = os.getenv("MAILGUN_API_KEY")
self.mailgun_domain = os.getenv("MAILGUN_DOMAIN") self.mailgun_domain = os.getenv("MAILGUN_DOMAIN")
self.mailgun_api_base = os.getenv("MAILGUN_API_BASE", "https://api.mailgun.net/v3") self.mailgun_api_base = os.getenv("MAILGUN_API_BASE", "https://api.mailgun.net/v3")
@ -32,8 +42,11 @@ class EmailService:
self.email_from = os.getenv("EMAIL_FROM", "noreply@elmeg.xyz") self.email_from = os.getenv("EMAIL_FROM", "noreply@elmeg.xyz")
self.frontend_url = os.getenv("FRONTEND_URL", "https://elmeg.xyz") self.frontend_url = os.getenv("FRONTEND_URL", "https://elmeg.xyz")
# Determine which provider to use # Determine which provider to use (priority: SMTP -> Mailgun -> SES -> Dummy)
if self.mailgun_api_key and self.mailgun_domain: if self.smtp_host and self.smtp_username and self.smtp_password:
self.provider = "smtp"
print(f"Email service: Using SMTP ({self.smtp_host}:{self.smtp_port})")
elif self.mailgun_api_key and self.mailgun_domain:
self.provider = "mailgun" self.provider = "mailgun"
print(f"Email service: Using Mailgun ({self.mailgun_domain})") print(f"Email service: Using Mailgun ({self.mailgun_domain})")
elif BOTO3_AVAILABLE and self.aws_access_key_id and self.aws_secret_access_key: elif BOTO3_AVAILABLE and self.aws_access_key_id and self.aws_secret_access_key:
@ -51,13 +64,45 @@ class EmailService:
def send_email(self, to_email: str, subject: str, html_content: str, text_content: str): def send_email(self, to_email: str, subject: str, html_content: str, text_content: str):
"""Send an email using configured provider""" """Send an email using configured provider"""
if self.provider == "mailgun": if self.provider == "smtp":
return self._send_smtp(to_email, subject, html_content, text_content)
elif self.provider == "mailgun":
return self._send_mailgun(to_email, subject, html_content, text_content) return self._send_mailgun(to_email, subject, html_content, text_content)
elif self.provider == "ses": elif self.provider == "ses":
return self._send_ses(to_email, subject, html_content, text_content) return self._send_ses(to_email, subject, html_content, text_content)
else: else:
return self._send_dummy(to_email, subject, text_content) return self._send_dummy(to_email, subject, text_content)
def _send_smtp(self, to_email: str, subject: str, html_content: str, text_content: str):
"""Send email via SMTP (Postal or any SMTP server)"""
try:
# Create message
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = f"Elmeg <{self.email_from}>"
msg["To"] = to_email
# Attach text and HTML parts
msg.attach(MIMEText(text_content, "plain"))
msg.attach(MIMEText(html_content, "html"))
# Connect and send
if self.smtp_use_tls:
server = smtplib.SMTP(self.smtp_host, self.smtp_port)
server.starttls()
else:
server = smtplib.SMTP(self.smtp_host, self.smtp_port)
server.login(self.smtp_username, self.smtp_password)
server.sendmail(self.email_from, to_email, msg.as_string())
server.quit()
print(f"Email sent via SMTP to {to_email}")
return True
except Exception as e:
print(f"Error sending email via SMTP: {e}")
return False
def _send_mailgun(self, to_email: str, subject: str, html_content: str, text_content: str): def _send_mailgun(self, to_email: str, subject: str, html_content: str, text_content: str):
"""Send email via Mailgun API""" """Send email via Mailgun API"""
try: try: