# app/modules/messaging/services/email_service.py
"""
Email service with multi-provider support.
Supports:
- SMTP (default)
- SendGrid
- Mailgun
- Amazon SES
Features:
- Multi-language templates from database
- Store template overrides
- Jinja2 template rendering
- Email logging and tracking
- Queue support via background tasks
- Branding based on store tier (whitelabel)
Language Resolution (priority order):
1. Explicit language parameter
2. Customer's preferred language (if customer context)
3. Store's storefront language
4. Platform default (en)
Template Resolution (priority order):
1. Store override (if store_id and template is not platform-only)
2. Platform template
3. English fallback (if requested language not found)
"""
import json
import logging
import smtplib
from abc import ABC, abstractmethod
from dataclasses import dataclass
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from typing import Any
from jinja2 import BaseLoader, Environment
from sqlalchemy.orm import Session
from app.core.config import settings
from app.modules.messaging.models import (
EmailLog,
EmailStatus,
EmailTemplate,
StoreEmailTemplate,
)
logger = logging.getLogger(__name__)
# Platform branding constants
PLATFORM_NAME = "Wizamart"
PLATFORM_SUPPORT_EMAIL = "support@wizamart.com"
PLATFORM_DEFAULT_LANGUAGE = "en"
SUPPORTED_LANGUAGES = ["en", "fr", "de", "lb"]
# Tiers that get white-label (no "Powered by Wizamart" footer)
WHITELABEL_TIERS = {"business", "enterprise"}
# Powered by Wizamart footer (added for Essential/Professional tiers)
POWERED_BY_FOOTER_HTML = """
"""
POWERED_BY_FOOTER_TEXT = "\n\n---\nPowered by Wizamart - https://wizamart.com"
@dataclass
class ResolvedTemplate:
"""Resolved template content after checking store overrides."""
subject: str
body_html: str
body_text: str | None
is_store_override: bool
template_id: int | None # Platform template ID (None if store override)
template_code: str
language: str
@dataclass
class BrandingContext:
"""Branding variables for email templates."""
platform_name: str
platform_logo_url: str | None
support_email: str
store_name: str | None
store_logo_url: str | None
is_whitelabel: bool
# =============================================================================
# EMAIL PROVIDER ABSTRACTION
# =============================================================================
class EmailProvider(ABC):
"""Abstract base class for email providers."""
@abstractmethod
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
"""
Send an email.
Returns:
tuple: (success, provider_message_id, error_message)
"""
class SMTPProvider(EmailProvider):
"""SMTP email provider."""
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
try:
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = f"{from_name} <{from_email}>" if from_name else from_email
msg["To"] = f"{to_name} <{to_email}>" if to_name else to_email
if reply_to:
msg["Reply-To"] = reply_to
# Attach text and HTML parts
if body_text:
msg.attach(MIMEText(body_text, "plain", "utf-8"))
msg.attach(MIMEText(body_html, "html", "utf-8"))
# Connect and send (10-second timeout to fail fast)
timeout = 10
if settings.smtp_use_ssl:
server = smtplib.SMTP_SSL(settings.smtp_host, settings.smtp_port, timeout=timeout)
else:
server = smtplib.SMTP(settings.smtp_host, settings.smtp_port, timeout=timeout)
try:
if settings.smtp_use_tls and not settings.smtp_use_ssl:
server.starttls()
if settings.smtp_user and settings.smtp_password:
server.login(settings.smtp_user, settings.smtp_password)
server.sendmail(from_email, [to_email], msg.as_string())
return True, None, None
finally:
server.quit()
except Exception as e:
logger.error(f"SMTP send error: {e}")
return False, None, str(e)
class SendGridProvider(EmailProvider):
"""SendGrid email provider."""
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
try:
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Content, Email, Mail, To
message = Mail(
from_email=Email(from_email, from_name),
to_emails=To(to_email, to_name),
subject=subject,
)
message.add_content(Content("text/html", body_html))
if body_text:
message.add_content(Content("text/plain", body_text))
if reply_to:
message.reply_to = Email(reply_to)
sg = SendGridAPIClient(settings.sendgrid_api_key)
response = sg.send(message)
if response.status_code in (200, 201, 202):
message_id = response.headers.get("X-Message-Id")
return True, message_id, None
return False, None, f"SendGrid error: {response.status_code}"
except ImportError:
return False, None, "SendGrid library not installed. Run: pip install sendgrid"
except Exception as e:
logger.error(f"SendGrid send error: {e}")
return False, None, str(e)
class MailgunProvider(EmailProvider):
"""Mailgun email provider."""
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
try:
import requests
from_str = f"{from_name} <{from_email}>" if from_name else from_email
to_str = f"{to_name} <{to_email}>" if to_name else to_email
data = {
"from": from_str,
"to": to_str,
"subject": subject,
"html": body_html,
}
if body_text:
data["text"] = body_text
if reply_to:
data["h:Reply-To"] = reply_to
response = requests.post(
f"https://api.mailgun.net/v3/{settings.mailgun_domain}/messages",
auth=("api", settings.mailgun_api_key),
data=data,
timeout=30,
)
if response.status_code == 200:
result = response.json()
return True, result.get("id"), None
return False, None, f"Mailgun error: {response.status_code} - {response.text}"
except Exception as e:
logger.error(f"Mailgun send error: {e}")
return False, None, str(e)
class SESProvider(EmailProvider):
"""Amazon SES email provider."""
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
try:
import boto3
ses = boto3.client(
"ses",
region_name=settings.aws_region,
aws_access_key_id=settings.aws_access_key_id,
aws_secret_access_key=settings.aws_secret_access_key,
)
from_str = f"{from_name} <{from_email}>" if from_name else from_email
body = {"Html": {"Charset": "UTF-8", "Data": body_html}}
if body_text:
body["Text"] = {"Charset": "UTF-8", "Data": body_text}
kwargs = {
"Source": from_str,
"Destination": {"ToAddresses": [to_email]},
"Message": {
"Subject": {"Charset": "UTF-8", "Data": subject},
"Body": body,
},
}
if reply_to:
kwargs["ReplyToAddresses"] = [reply_to]
response = ses.send_email(**kwargs)
return True, response.get("MessageId"), None
except ImportError:
return False, None, "boto3 library not installed. Run: pip install boto3"
except Exception as e:
logger.error(f"SES send error: {e}")
return False, None, str(e)
class DebugProvider(EmailProvider):
"""Debug provider - logs emails instead of sending."""
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
logger.info(
f"\n{'='*60}\n"
f"DEBUG EMAIL\n"
f"{'='*60}\n"
f"To: {to_name} <{to_email}>\n"
f"From: {from_name} <{from_email}>\n"
f"Reply-To: {reply_to}\n"
f"Subject: {subject}\n"
f"{'='*60}\n"
f"Body (text):\n{body_text or '(none)'}\n"
f"{'='*60}\n"
f"Body (html):\n{body_html[:500]}...\n"
f"{'='*60}\n"
)
return True, f"debug-{to_email}", None
# =============================================================================
# PLATFORM CONFIG HELPERS (DB overrides .env)
# =============================================================================
def get_platform_email_config(db: Session) -> dict:
"""
Get effective platform email configuration.
Priority: Database settings > Environment variables (.env)
Returns:
Dictionary with all email configuration values
"""
from app.modules.tenancy.models import AdminSetting
def get_db_setting(key: str) -> str | None:
setting = db.query(AdminSetting).filter(AdminSetting.key == key).first()
return setting.value if setting else None
config = {}
# Provider
db_provider = get_db_setting("email_provider")
config["provider"] = db_provider if db_provider else settings.email_provider
# From settings
db_from_email = get_db_setting("email_from_address")
config["from_email"] = db_from_email if db_from_email else settings.email_from_address
db_from_name = get_db_setting("email_from_name")
config["from_name"] = db_from_name if db_from_name else settings.email_from_name
db_reply_to = get_db_setting("email_reply_to")
config["reply_to"] = db_reply_to if db_reply_to else settings.email_reply_to
# SMTP settings
db_smtp_host = get_db_setting("smtp_host")
config["smtp_host"] = db_smtp_host if db_smtp_host else settings.smtp_host
db_smtp_port = get_db_setting("smtp_port")
config["smtp_port"] = int(db_smtp_port) if db_smtp_port else settings.smtp_port
db_smtp_user = get_db_setting("smtp_user")
config["smtp_user"] = db_smtp_user if db_smtp_user else settings.smtp_user
db_smtp_password = get_db_setting("smtp_password")
config["smtp_password"] = db_smtp_password if db_smtp_password else settings.smtp_password
db_smtp_use_tls = get_db_setting("smtp_use_tls")
config["smtp_use_tls"] = db_smtp_use_tls.lower() in ("true", "1", "yes") if db_smtp_use_tls else settings.smtp_use_tls
db_smtp_use_ssl = get_db_setting("smtp_use_ssl")
config["smtp_use_ssl"] = db_smtp_use_ssl.lower() in ("true", "1", "yes") if db_smtp_use_ssl else settings.smtp_use_ssl
# SendGrid
db_sendgrid_key = get_db_setting("sendgrid_api_key")
config["sendgrid_api_key"] = db_sendgrid_key if db_sendgrid_key else settings.sendgrid_api_key
# Mailgun
db_mailgun_key = get_db_setting("mailgun_api_key")
config["mailgun_api_key"] = db_mailgun_key if db_mailgun_key else settings.mailgun_api_key
db_mailgun_domain = get_db_setting("mailgun_domain")
config["mailgun_domain"] = db_mailgun_domain if db_mailgun_domain else settings.mailgun_domain
# AWS SES
db_aws_key = get_db_setting("aws_access_key_id")
config["aws_access_key_id"] = db_aws_key if db_aws_key else settings.aws_access_key_id
db_aws_secret = get_db_setting("aws_secret_access_key")
config["aws_secret_access_key"] = db_aws_secret if db_aws_secret else settings.aws_secret_access_key
db_aws_region = get_db_setting("aws_region")
config["aws_region"] = db_aws_region if db_aws_region else settings.aws_region
# Behavior
db_enabled = get_db_setting("email_enabled")
config["enabled"] = db_enabled.lower() in ("true", "1", "yes") if db_enabled else settings.email_enabled
db_debug = get_db_setting("email_debug")
config["debug"] = db_debug.lower() in ("true", "1", "yes") if db_debug else settings.email_debug
return config
# =============================================================================
# CONFIGURABLE PLATFORM PROVIDERS (use config dict instead of global settings)
# =============================================================================
class ConfigurableSMTPProvider(EmailProvider):
"""SMTP provider using config dictionary."""
def __init__(self, config: dict):
self.config = config
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
try:
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = f"{from_name} <{from_email}>" if from_name else from_email
msg["To"] = f"{to_name} <{to_email}>" if to_name else to_email
if reply_to:
msg["Reply-To"] = reply_to
if body_text:
msg.attach(MIMEText(body_text, "plain", "utf-8"))
msg.attach(MIMEText(body_html, "html", "utf-8"))
# Use 10-second timeout to fail fast on bad SMTP settings
timeout = 10
if self.config.get("smtp_use_ssl"):
server = smtplib.SMTP_SSL(self.config["smtp_host"], self.config["smtp_port"], timeout=timeout)
else:
server = smtplib.SMTP(self.config["smtp_host"], self.config["smtp_port"], timeout=timeout)
try:
if self.config.get("smtp_use_tls") and not self.config.get("smtp_use_ssl"):
server.starttls()
if self.config.get("smtp_user") and self.config.get("smtp_password"):
server.login(self.config["smtp_user"], self.config["smtp_password"])
server.sendmail(from_email, [to_email], msg.as_string())
return True, None, None
finally:
server.quit()
except Exception as e:
logger.error(f"Configurable SMTP send error: {e}")
return False, None, str(e)
class ConfigurableSendGridProvider(EmailProvider):
"""SendGrid provider using config dictionary."""
def __init__(self, config: dict):
self.config = config
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
try:
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Content, Email, Mail, To
message = Mail(
from_email=Email(from_email, from_name),
to_emails=To(to_email, to_name),
subject=subject,
)
message.add_content(Content("text/html", body_html))
if body_text:
message.add_content(Content("text/plain", body_text))
if reply_to:
message.reply_to = Email(reply_to)
sg = SendGridAPIClient(self.config["sendgrid_api_key"])
response = sg.send(message)
if response.status_code in (200, 201, 202):
message_id = response.headers.get("X-Message-Id")
return True, message_id, None
return False, None, f"SendGrid error: {response.status_code}"
except ImportError:
return False, None, "SendGrid library not installed"
except Exception as e:
logger.error(f"Configurable SendGrid send error: {e}")
return False, None, str(e)
class ConfigurableMailgunProvider(EmailProvider):
"""Mailgun provider using config dictionary."""
def __init__(self, config: dict):
self.config = config
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
try:
import requests
from_str = f"{from_name} <{from_email}>" if from_name else from_email
to_str = f"{to_name} <{to_email}>" if to_name else to_email
data = {
"from": from_str,
"to": to_str,
"subject": subject,
"html": body_html,
}
if body_text:
data["text"] = body_text
if reply_to:
data["h:Reply-To"] = reply_to
response = requests.post(
f"https://api.mailgun.net/v3/{self.config['mailgun_domain']}/messages",
auth=("api", self.config["mailgun_api_key"]),
data=data,
timeout=30,
)
if response.status_code == 200:
result = response.json()
return True, result.get("id"), None
return False, None, f"Mailgun error: {response.status_code} - {response.text}"
except Exception as e:
logger.error(f"Configurable Mailgun send error: {e}")
return False, None, str(e)
class ConfigurableSESProvider(EmailProvider):
"""Amazon SES provider using config dictionary."""
def __init__(self, config: dict):
self.config = config
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
try:
import boto3
ses = boto3.client(
"ses",
region_name=self.config["aws_region"],
aws_access_key_id=self.config["aws_access_key_id"],
aws_secret_access_key=self.config["aws_secret_access_key"],
)
from_str = f"{from_name} <{from_email}>" if from_name else from_email
body = {"Html": {"Charset": "UTF-8", "Data": body_html}}
if body_text:
body["Text"] = {"Charset": "UTF-8", "Data": body_text}
kwargs = {
"Source": from_str,
"Destination": {"ToAddresses": [to_email]},
"Message": {
"Subject": {"Charset": "UTF-8", "Data": subject},
"Body": body,
},
}
if reply_to:
kwargs["ReplyToAddresses"] = [reply_to]
response = ses.send_email(**kwargs)
return True, response.get("MessageId"), None
except ImportError:
return False, None, "boto3 library not installed"
except Exception as e:
logger.error(f"Configurable SES send error: {e}")
return False, None, str(e)
def get_platform_provider(db: Session) -> EmailProvider:
"""
Get the configured email provider using effective platform config.
Uses database settings if available, otherwise falls back to .env.
"""
config = get_platform_email_config(db)
if config.get("debug"):
return DebugProvider()
provider_map = {
"smtp": ConfigurableSMTPProvider,
"sendgrid": ConfigurableSendGridProvider,
"mailgun": ConfigurableMailgunProvider,
"ses": ConfigurableSESProvider,
}
provider_name = config.get("provider", "smtp").lower()
provider_class = provider_map.get(provider_name)
if not provider_class:
logger.warning(f"Unknown email provider: {provider_name}, using SMTP")
return ConfigurableSMTPProvider(config)
return provider_class(config)
# =============================================================================
# STORE EMAIL PROVIDERS
# =============================================================================
class StoreSMTPProvider(EmailProvider):
"""SMTP provider using store-specific settings."""
def __init__(self, store_settings):
self.settings = store_settings
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
try:
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = f"{from_name} <{from_email}>" if from_name else from_email
msg["To"] = f"{to_name} <{to_email}>" if to_name else to_email
if reply_to:
msg["Reply-To"] = reply_to
if body_text:
msg.attach(MIMEText(body_text, "plain", "utf-8"))
msg.attach(MIMEText(body_html, "html", "utf-8"))
# Use store's SMTP settings (10-second timeout to fail fast)
timeout = 10
if self.settings.smtp_use_ssl:
server = smtplib.SMTP_SSL(self.settings.smtp_host, self.settings.smtp_port, timeout=timeout)
else:
server = smtplib.SMTP(self.settings.smtp_host, self.settings.smtp_port, timeout=timeout)
try:
if self.settings.smtp_use_tls and not self.settings.smtp_use_ssl:
server.starttls()
if self.settings.smtp_username and self.settings.smtp_password:
server.login(self.settings.smtp_username, self.settings.smtp_password)
server.sendmail(from_email, [to_email], msg.as_string())
return True, None, None
finally:
server.quit()
except Exception as e:
logger.error(f"Store SMTP send error: {e}")
return False, None, str(e)
class StoreSendGridProvider(EmailProvider):
"""SendGrid provider using store-specific API key."""
def __init__(self, store_settings):
self.settings = store_settings
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
try:
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Content, Email, Mail, To
message = Mail(
from_email=Email(from_email, from_name),
to_emails=To(to_email, to_name),
subject=subject,
)
message.add_content(Content("text/html", body_html))
if body_text:
message.add_content(Content("text/plain", body_text))
if reply_to:
message.reply_to = Email(reply_to)
sg = SendGridAPIClient(self.settings.sendgrid_api_key)
response = sg.send(message)
if response.status_code in (200, 201, 202):
message_id = response.headers.get("X-Message-Id")
return True, message_id, None
return False, None, f"SendGrid error: {response.status_code}"
except ImportError:
return False, None, "SendGrid library not installed"
except Exception as e:
logger.error(f"Store SendGrid send error: {e}")
return False, None, str(e)
class StoreMailgunProvider(EmailProvider):
"""Mailgun provider using store-specific settings."""
def __init__(self, store_settings):
self.settings = store_settings
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
try:
import requests
from_str = f"{from_name} <{from_email}>" if from_name else from_email
to_str = f"{to_name} <{to_email}>" if to_name else to_email
data = {
"from": from_str,
"to": to_str,
"subject": subject,
"html": body_html,
}
if body_text:
data["text"] = body_text
if reply_to:
data["h:Reply-To"] = reply_to
response = requests.post(
f"https://api.mailgun.net/v3/{self.settings.mailgun_domain}/messages",
auth=("api", self.settings.mailgun_api_key),
data=data,
timeout=30,
)
if response.status_code == 200:
result = response.json()
return True, result.get("id"), None
return False, None, f"Mailgun error: {response.status_code} - {response.text}"
except Exception as e:
logger.error(f"Store Mailgun send error: {e}")
return False, None, str(e)
class StoreSESProvider(EmailProvider):
"""Amazon SES provider using store-specific credentials."""
def __init__(self, store_settings):
self.settings = store_settings
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
try:
import boto3
ses = boto3.client(
"ses",
region_name=self.settings.ses_region,
aws_access_key_id=self.settings.ses_access_key_id,
aws_secret_access_key=self.settings.ses_secret_access_key,
)
from_str = f"{from_name} <{from_email}>" if from_name else from_email
body = {"Html": {"Charset": "UTF-8", "Data": body_html}}
if body_text:
body["Text"] = {"Charset": "UTF-8", "Data": body_text}
kwargs = {
"Source": from_str,
"Destination": {"ToAddresses": [to_email]},
"Message": {
"Subject": {"Charset": "UTF-8", "Data": subject},
"Body": body,
},
}
if reply_to:
kwargs["ReplyToAddresses"] = [reply_to]
response = ses.send_email(**kwargs)
return True, response.get("MessageId"), None
except ImportError:
return False, None, "boto3 library not installed"
except Exception as e:
logger.error(f"Store SES send error: {e}")
return False, None, str(e)
def get_store_provider(store_settings) -> EmailProvider | None:
"""
Create an email provider instance using store's settings.
Args:
store_settings: StoreEmailSettings model instance
Returns:
EmailProvider instance or None if not configured
"""
if not store_settings or not store_settings.is_configured:
return None
provider_map = {
"smtp": StoreSMTPProvider,
"sendgrid": StoreSendGridProvider,
"mailgun": StoreMailgunProvider,
"ses": StoreSESProvider,
}
provider_class = provider_map.get(store_settings.provider)
if not provider_class:
logger.warning(f"Unknown store email provider: {store_settings.provider}")
return None
return provider_class(store_settings)
# =============================================================================
# PLATFORM EMAIL PROVIDER
# =============================================================================
def get_provider() -> EmailProvider:
"""Get the configured email provider."""
if settings.email_debug:
return DebugProvider()
provider_map = {
"smtp": SMTPProvider,
"sendgrid": SendGridProvider,
"mailgun": MailgunProvider,
"ses": SESProvider,
}
provider_class = provider_map.get(settings.email_provider.lower())
if not provider_class:
logger.warning(f"Unknown email provider: {settings.email_provider}, using SMTP")
return SMTPProvider()
return provider_class()
class EmailService:
"""
Email service for sending templated emails.
Usage:
email_service = EmailService(db)
# Send using database template with store override support
email_service.send_template(
template_code="signup_welcome",
to_email="user@example.com",
to_name="John Doe",
variables={"first_name": "John", "login_url": "https://..."},
store_id=1,
# Language is resolved automatically from store/customer settings
)
# Send raw email
email_service.send_raw(
to_email="user@example.com",
subject="Hello",
body_html="Hello
",
)
Platform email configuration is loaded from:
1. Database (admin_settings table) - if settings exist
2. Environment variables (.env) - fallback
"""
def __init__(self, db: Session):
self.db = db
# Use configurable provider that checks DB first, then .env
self.provider = get_platform_provider(db)
# Cache the platform config for use in send_raw
self._platform_config = get_platform_email_config(db)
self.jinja_env = Environment(loader=BaseLoader())
# Cache store and feature data to avoid repeated queries
self._store_cache: dict[int, Any] = {}
self._feature_cache: dict[int, set[str]] = {}
self._store_email_settings_cache: dict[int, Any] = {}
self._store_tier_cache: dict[int, str | None] = {}
def _get_store(self, store_id: int):
"""Get store with caching."""
if store_id not in self._store_cache:
from app.modules.tenancy.models import Store
self._store_cache[store_id] = (
self.db.query(Store).filter(Store.id == store_id).first()
)
return self._store_cache[store_id]
def _has_feature(self, store_id: int, feature_code: str) -> bool:
"""Check if store has a specific feature enabled."""
if store_id not in self._feature_cache:
from app.modules.billing.services.feature_service import feature_service
try:
features = feature_service.get_store_features(self.db, store_id)
# Convert to set of feature codes
self._feature_cache[store_id] = {f.code for f in features.features}
except Exception:
self._feature_cache[store_id] = set()
return feature_code in self._feature_cache[store_id]
def _get_store_email_settings(self, store_id: int):
"""Get store email settings with caching."""
if store_id not in self._store_email_settings_cache:
from app.modules.messaging.models import StoreEmailSettings
self._store_email_settings_cache[store_id] = (
self.db.query(StoreEmailSettings)
.filter(StoreEmailSettings.store_id == store_id)
.first()
)
return self._store_email_settings_cache[store_id]
def _get_store_tier(self, store_id: int) -> str | None:
"""Get store's subscription tier with caching."""
if store_id not in self._store_tier_cache:
from app.modules.billing.services.subscription_service import (
subscription_service,
)
tier = subscription_service.get_current_tier(self.db, store_id)
self._store_tier_cache[store_id] = tier.value if tier else None
return self._store_tier_cache[store_id]
def _should_add_powered_by_footer(self, store_id: int | None) -> bool:
"""
Check if "Powered by Wizamart" footer should be added.
Footer is added for Essential and Professional tiers.
Business and Enterprise tiers get white-label (no footer).
"""
if not store_id:
return False # Platform emails don't get the footer
tier = self._get_store_tier(store_id)
if not tier:
return True # No tier = show footer (shouldn't happen normally)
return tier.lower() not in WHITELABEL_TIERS
def _inject_powered_by_footer(
self,
body_html: str,
body_text: str | None,
store_id: int | None,
) -> tuple[str, str | None]:
"""
Inject "Powered by Wizamart" footer if needed based on tier.
Returns:
Tuple of (modified_html, modified_text)
"""
if not self._should_add_powered_by_footer(store_id):
return body_html, body_text
# Inject footer before closing tag if present, otherwise append
if "" in body_html.lower():
# Find case-insensitively and inject before it
import re
body_html = re.sub(
r"()",
f"{POWERED_BY_FOOTER_HTML}\\1",
body_html,
flags=re.IGNORECASE,
)
else:
body_html += POWERED_BY_FOOTER_HTML
if body_text:
body_text += POWERED_BY_FOOTER_TEXT
return body_html, body_text
def resolve_language(
self,
explicit_language: str | None = None,
store_id: int | None = None,
customer_id: int | None = None,
) -> str:
"""
Resolve the language for an email.
Priority order:
1. Explicit language parameter
2. Customer's preferred language (if customer_id provided)
3. Store's storefront language (if store_id provided)
4. Platform default (en)
Args:
explicit_language: Explicitly requested language
store_id: Store ID for storefront language lookup
customer_id: Customer ID for preferred language lookup
Returns:
Resolved language code (one of: en, fr, de, lb)
"""
# 1. Explicit language takes priority
if explicit_language and explicit_language in SUPPORTED_LANGUAGES:
return explicit_language
# 2. Customer's preferred language
if customer_id:
from app.modules.customers.models.customer import Customer
customer = (
self.db.query(Customer).filter(Customer.id == customer_id).first()
)
if customer and customer.preferred_language in SUPPORTED_LANGUAGES:
return customer.preferred_language
# 3. Store's storefront language
if store_id:
store = self._get_store(store_id)
if store and store.storefront_language in SUPPORTED_LANGUAGES:
return store.storefront_language
# 4. Platform default
return PLATFORM_DEFAULT_LANGUAGE
def get_branding(self, store_id: int | None = None) -> BrandingContext:
"""
Get branding context for email templates.
If store has white_label feature enabled (Enterprise tier),
platform branding is replaced with store branding.
Args:
store_id: Optional store ID
Returns:
BrandingContext with appropriate branding variables
"""
store = None
is_whitelabel = False
if store_id:
store = self._get_store(store_id)
is_whitelabel = self._has_feature(store_id, "white_label")
if is_whitelabel and store:
# Whitelabel: use store branding throughout
return BrandingContext(
platform_name=store.name,
platform_logo_url=store.get_logo_url(),
support_email=store.support_email or PLATFORM_SUPPORT_EMAIL,
store_name=store.name,
store_logo_url=store.get_logo_url(),
is_whitelabel=True,
)
# Standard: Wizamart branding with store details
return BrandingContext(
platform_name=PLATFORM_NAME,
platform_logo_url=None, # Use default platform logo
support_email=PLATFORM_SUPPORT_EMAIL,
store_name=store.name if store else None,
store_logo_url=store.get_logo_url() if store else None,
is_whitelabel=False,
)
def resolve_template(
self,
template_code: str,
language: str,
store_id: int | None = None,
) -> ResolvedTemplate | None:
"""
Resolve template content with store override support.
Resolution order:
1. Check for store override (if store_id and template is not platform-only)
2. Fall back to platform template
3. Fall back to English if language not found
Args:
template_code: Template code (e.g., "password_reset")
language: Language code
store_id: Optional store ID for override lookup
Returns:
ResolvedTemplate with content, or None if not found
"""
# First, get platform template to check if it's platform-only
platform_template = self.get_template(template_code, language)
if not platform_template:
logger.warning(f"Template not found: {template_code} ({language})")
return None
# Check for store override (if not platform-only)
if store_id and not platform_template.is_platform_only:
store_override = StoreEmailTemplate.get_override(
self.db, store_id, template_code, language
)
if store_override:
return ResolvedTemplate(
subject=store_override.subject,
body_html=store_override.body_html,
body_text=store_override.body_text,
is_store_override=True,
template_id=None,
template_code=template_code,
language=language,
)
# Use platform template
return ResolvedTemplate(
subject=platform_template.subject,
body_html=platform_template.body_html,
body_text=platform_template.body_text,
is_store_override=False,
template_id=platform_template.id,
template_code=template_code,
language=language,
)
def get_template(
self, template_code: str, language: str = "en"
) -> EmailTemplate | None:
"""Get platform email template from database with fallback to English."""
template = (
self.db.query(EmailTemplate)
.filter(
EmailTemplate.code == template_code,
EmailTemplate.language == language,
EmailTemplate.is_active == True, # noqa: E712
)
.first()
)
# Fallback to English if not found
if not template and language != "en":
template = (
self.db.query(EmailTemplate)
.filter(
EmailTemplate.code == template_code,
EmailTemplate.language == "en",
EmailTemplate.is_active == True, # noqa: E712
)
.first()
)
return template
def render_template(self, template_string: str, variables: dict[str, Any]) -> str:
"""Render a Jinja2 template string with variables."""
try:
template = self.jinja_env.from_string(template_string)
return template.render(**variables)
except Exception as e:
logger.error(f"Template rendering error: {e}")
return template_string
def send_template(
self,
template_code: str,
to_email: str,
to_name: str | None = None,
language: str | None = None,
variables: dict[str, Any] | None = None,
store_id: int | None = None,
customer_id: int | None = None,
user_id: int | None = None,
related_type: str | None = None,
related_id: int | None = None,
include_branding: bool = True,
) -> EmailLog:
"""
Send an email using a database template with store override support.
Args:
template_code: Template code (e.g., "signup_welcome")
to_email: Recipient email address
to_name: Recipient name (optional)
language: Language code (auto-resolved if None)
variables: Template variables dict
store_id: Store ID for override lookup and logging
customer_id: Customer ID for language resolution
user_id: Related user ID for logging
related_type: Related entity type (e.g., "order")
related_id: Related entity ID
include_branding: Whether to inject branding variables (default: True)
Returns:
EmailLog record
"""
variables = variables or {}
# Resolve language (uses customer -> store -> platform default order)
resolved_language = self.resolve_language(
explicit_language=language,
store_id=store_id,
customer_id=customer_id,
)
# Resolve template (checks store override, falls back to platform)
resolved = self.resolve_template(template_code, resolved_language, store_id)
if not resolved:
logger.error(f"Email template not found: {template_code} ({resolved_language})")
# Create failed log entry
log = EmailLog(
template_code=template_code,
recipient_email=to_email,
recipient_name=to_name,
subject=f"[Template not found: {template_code}]",
from_email=settings.email_from_address,
from_name=settings.email_from_name,
status=EmailStatus.FAILED.value,
error_message=f"Template not found: {template_code} ({resolved_language})",
provider=settings.email_provider,
store_id=store_id,
user_id=user_id,
related_type=related_type,
related_id=related_id,
)
self.db.add(log)
self.db.commit() # noqa: SVC-006 - Email logs are side effects, commit immediately
return log
# Inject branding variables if requested
if include_branding:
branding = self.get_branding(store_id)
variables = {
**variables,
"platform_name": branding.platform_name,
"platform_logo_url": branding.platform_logo_url,
"support_email": branding.support_email,
"store_name": branding.store_name,
"store_logo_url": branding.store_logo_url,
"is_whitelabel": branding.is_whitelabel,
}
# Render template
subject = self.render_template(resolved.subject, variables)
body_html = self.render_template(resolved.body_html, variables)
body_text = (
self.render_template(resolved.body_text, variables)
if resolved.body_text
else None
)
return self.send_raw(
to_email=to_email,
to_name=to_name,
subject=subject,
body_html=body_html,
body_text=body_text,
template_code=template_code,
template_id=resolved.template_id,
store_id=store_id,
user_id=user_id,
related_type=related_type,
related_id=related_id,
extra_data=json.dumps(variables) if variables else None,
)
def send_raw(
self,
to_email: str,
subject: str,
body_html: str,
to_name: str | None = None,
body_text: str | None = None,
from_email: str | None = None,
from_name: str | None = None,
reply_to: str | None = None,
template_code: str | None = None,
template_id: int | None = None,
store_id: int | None = None,
user_id: int | None = None,
related_type: str | None = None,
related_id: int | None = None,
extra_data: str | None = None,
is_platform_email: bool = False,
) -> EmailLog:
"""
Send a raw email without using a template.
For store emails (when store_id is provided and is_platform_email=False):
- Uses store's SMTP/provider settings if configured
- Uses store's from_email, from_name, reply_to
- Adds "Powered by Wizamart" footer for Essential/Professional tiers
For platform emails (is_platform_email=True or no store_id):
- Uses platform's email settings from config
- No "Powered by Wizamart" footer
Args:
is_platform_email: If True, always use platform settings (for billing, etc.)
Returns:
EmailLog record
"""
# Determine which provider and settings to use
store_settings = None
store_provider = None
provider_name = self._platform_config.get("provider", settings.email_provider)
if store_id and not is_platform_email:
store_settings = self._get_store_email_settings(store_id)
if store_settings and store_settings.is_configured:
store_provider = get_store_provider(store_settings)
if store_provider:
# Use store's email identity
from_email = from_email or store_settings.from_email
from_name = from_name or store_settings.from_name
reply_to = reply_to or store_settings.reply_to_email
provider_name = f"store_{store_settings.provider}"
logger.debug(f"Using store email provider: {store_settings.provider}")
# Fall back to platform settings if no store provider
# Uses DB config if available, otherwise .env
if not store_provider:
from_email = from_email or self._platform_config.get("from_email", settings.email_from_address)
from_name = from_name or self._platform_config.get("from_name", settings.email_from_name)
reply_to = reply_to or self._platform_config.get("reply_to") or settings.email_reply_to or None
# Inject "Powered by Wizamart" footer for non-whitelabel tiers
if store_id and not is_platform_email:
body_html, body_text = self._inject_powered_by_footer(
body_html, body_text, store_id
)
# Create log entry
log = EmailLog(
template_code=template_code,
template_id=template_id,
recipient_email=to_email,
recipient_name=to_name,
subject=subject,
body_html=body_html,
body_text=body_text,
from_email=from_email,
from_name=from_name,
reply_to=reply_to,
status=EmailStatus.PENDING.value,
provider=provider_name,
store_id=store_id,
user_id=user_id,
related_type=related_type,
related_id=related_id,
extra_data=extra_data,
)
self.db.add(log)
self.db.flush()
# Check if emails are disabled (uses DB config if available)
email_enabled = self._platform_config.get("enabled", settings.email_enabled)
if not email_enabled:
log.status = EmailStatus.FAILED.value
log.error_message = "Email sending is disabled"
self.db.commit() # noqa: SVC-006 - Email logs are side effects, commit immediately
logger.info(f"Email sending disabled, skipping: {to_email}")
return log
# Use store provider if available, otherwise platform provider
provider_to_use = store_provider or self.provider
# Send email
success, message_id, error = provider_to_use.send(
to_email=to_email,
to_name=to_name,
subject=subject,
body_html=body_html,
body_text=body_text,
from_email=from_email,
from_name=from_name,
reply_to=reply_to,
)
if success:
log.mark_sent(message_id)
logger.info(f"Email sent to {to_email}: {subject} (via {provider_name})")
else:
log.mark_failed(error or "Unknown error")
logger.error(f"Email failed to {to_email}: {error}")
self.db.commit() # noqa: SVC-006 - Email logs are side effects, commit immediately
return log
# =============================================================================
# CONVENIENCE FUNCTIONS
# =============================================================================
def send_email(
db: Session,
template_code: str,
to_email: str,
to_name: str | None = None,
language: str | None = None,
variables: dict[str, Any] | None = None,
store_id: int | None = None,
customer_id: int | None = None,
**kwargs,
) -> EmailLog:
"""
Convenience function to send a templated email.
Args:
db: Database session
template_code: Template code (e.g., "password_reset")
to_email: Recipient email address
to_name: Recipient name (optional)
language: Language code (auto-resolved from customer/store if None)
variables: Template variables dict
store_id: Store ID for override lookup and branding
customer_id: Customer ID for language resolution
**kwargs: Additional arguments passed to send_template
Returns:
EmailLog record
"""
service = EmailService(db)
return service.send_template(
template_code=template_code,
to_email=to_email,
to_name=to_name,
language=language,
variables=variables,
store_id=store_id,
customer_id=customer_id,
**kwargs,
)