# app/services/vendor_email_settings_service.py
"""
Vendor Email Settings Service.
Handles CRUD operations for vendor email configuration:
- SMTP settings
- Advanced providers (SendGrid, Mailgun, SES) - tier-gated
- Sender identity (from_email, from_name, reply_to)
- Signature/footer customization
- Configuration verification via test email
"""
import logging
import smtplib
from datetime import UTC, datetime
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from sqlalchemy.orm import Session
from app.exceptions import NotFoundError, ValidationError, AuthorizationError
from models.database import (
Vendor,
VendorEmailSettings,
EmailProvider,
PREMIUM_EMAIL_PROVIDERS,
VendorSubscription,
TierCode,
)
logger = logging.getLogger(__name__)
# Tiers that allow premium email providers
PREMIUM_TIERS = {TierCode.BUSINESS, TierCode.ENTERPRISE}
class VendorEmailSettingsService:
"""Service for managing vendor email settings."""
def __init__(self, db: Session):
self.db = db
# =========================================================================
# READ OPERATIONS
# =========================================================================
def get_settings(self, vendor_id: int) -> VendorEmailSettings | None:
"""Get email settings for a vendor."""
return (
self.db.query(VendorEmailSettings)
.filter(VendorEmailSettings.vendor_id == vendor_id)
.first()
)
def get_settings_or_404(self, vendor_id: int) -> VendorEmailSettings:
"""Get email settings or raise 404."""
settings = self.get_settings(vendor_id)
if not settings:
raise NotFoundError(
f"Email settings not found for vendor {vendor_id}. "
"Configure email settings to send emails."
)
return settings
def is_configured(self, vendor_id: int) -> bool:
"""Check if vendor has configured email settings."""
settings = self.get_settings(vendor_id)
return settings is not None and settings.is_configured
def get_status(self, vendor_id: int) -> dict:
"""
Get email configuration status for a vendor.
Returns:
dict with is_configured, is_verified, provider, etc.
"""
settings = self.get_settings(vendor_id)
if not settings:
return {
"is_configured": False,
"is_verified": False,
"provider": None,
"from_email": None,
"from_name": None,
"message": "Email settings not configured. Configure SMTP to send emails.",
}
return {
"is_configured": settings.is_configured,
"is_verified": settings.is_verified,
"provider": settings.provider,
"from_email": settings.from_email,
"from_name": settings.from_name,
"last_verified_at": settings.last_verified_at.isoformat() if settings.last_verified_at else None,
"verification_error": settings.verification_error,
"message": self._get_status_message(settings),
}
def _get_status_message(self, settings: VendorEmailSettings) -> str:
"""Generate a human-readable status message."""
if not settings.is_configured:
return "Complete your email configuration to send emails."
if not settings.is_verified:
return "Email configured but not verified. Send a test email to verify."
return "Email settings configured and verified."
# =========================================================================
# WRITE OPERATIONS
# =========================================================================
def create_or_update(
self,
vendor_id: int,
data: dict,
current_tier: TierCode | None = None,
) -> VendorEmailSettings:
"""
Create or update vendor email settings.
Args:
vendor_id: Vendor ID
data: Settings data (from_email, from_name, smtp_*, etc.)
current_tier: Vendor's current subscription tier (for premium provider validation)
Returns:
Updated VendorEmailSettings
"""
# Validate premium provider access
provider = data.get("provider", "smtp")
if provider in [p.value for p in PREMIUM_EMAIL_PROVIDERS]:
if current_tier not in PREMIUM_TIERS:
raise AuthorizationError(
f"Provider '{provider}' requires Business or Enterprise tier. "
"Upgrade your plan to use advanced email providers."
)
settings = self.get_settings(vendor_id)
if not settings:
settings = VendorEmailSettings(vendor_id=vendor_id)
self.db.add(settings)
# Update fields
for field in [
"from_email",
"from_name",
"reply_to_email",
"signature_text",
"signature_html",
"provider",
# SMTP
"smtp_host",
"smtp_port",
"smtp_username",
"smtp_password",
"smtp_use_tls",
"smtp_use_ssl",
# SendGrid
"sendgrid_api_key",
# Mailgun
"mailgun_api_key",
"mailgun_domain",
# SES
"ses_access_key_id",
"ses_secret_access_key",
"ses_region",
]:
if field in data and data[field] is not None:
# Don't overwrite passwords/keys with empty strings
if field.endswith(("_password", "_key", "_access_key")) and data[field] == "":
continue
setattr(settings, field, data[field])
# Update configuration status
settings.update_configuration_status()
# Reset verification if provider/credentials changed
if any(
f in data
for f in ["provider", "smtp_host", "smtp_password", "sendgrid_api_key", "mailgun_api_key", "ses_access_key_id"]
):
settings.is_verified = False
settings.verification_error = None
self.db.commit()
self.db.refresh(settings)
logger.info(f"Updated email settings for vendor {vendor_id}: provider={settings.provider}")
return settings
def delete(self, vendor_id: int) -> bool:
"""Delete email settings for a vendor."""
settings = self.get_settings(vendor_id)
if settings:
self.db.delete(settings)
self.db.commit()
logger.info(f"Deleted email settings for vendor {vendor_id}")
return True
return False
# =========================================================================
# VERIFICATION
# =========================================================================
def verify_settings(self, vendor_id: int, test_email: str) -> dict:
"""
Verify email settings by sending a test email.
Args:
vendor_id: Vendor ID
test_email: Email address to send test email to
Returns:
dict with success status and message
"""
settings = self.get_settings_or_404(vendor_id)
if not settings.is_fully_configured():
raise ValidationError("Email settings incomplete. Configure all required fields first.")
try:
# Send test email based on provider
if settings.provider == EmailProvider.SMTP.value:
self._send_smtp_test(settings, test_email)
elif settings.provider == EmailProvider.SENDGRID.value:
self._send_sendgrid_test(settings, test_email)
elif settings.provider == EmailProvider.MAILGUN.value:
self._send_mailgun_test(settings, test_email)
elif settings.provider == EmailProvider.SES.value:
self._send_ses_test(settings, test_email)
else:
raise ValidationError(f"Unknown provider: {settings.provider}")
# Mark as verified
settings.mark_verified()
self.db.commit()
logger.info(f"Email settings verified for vendor {vendor_id}")
return {
"success": True,
"message": f"Test email sent successfully to {test_email}",
}
except Exception as e:
error_msg = str(e)
settings.mark_verification_failed(error_msg)
self.db.commit()
logger.warning(f"Email verification failed for vendor {vendor_id}: {error_msg}")
return {
"success": False,
"message": f"Failed to send test email: {error_msg}",
}
def _send_smtp_test(self, settings: VendorEmailSettings, to_email: str) -> None:
"""Send test email via SMTP."""
msg = MIMEMultipart("alternative")
msg["Subject"] = "Wizamart Email Configuration Test"
msg["From"] = f"{settings.from_name} <{settings.from_email}>"
msg["To"] = to_email
text_content = (
"This is a test email from Wizamart.\n\n"
"Your email settings are configured correctly!\n\n"
f"Provider: SMTP\n"
f"Host: {settings.smtp_host}\n"
)
html_content = f"""
Email Configuration Test
This is a test email from Wizamart.
Your email settings are configured correctly!
Provider: SMTP
Host: {settings.smtp_host}
Sent at: {datetime.now(UTC).strftime('%Y-%m-%d %H:%M:%S UTC')}
"""
msg.attach(MIMEText(text_content, "plain"))
msg.attach(MIMEText(html_content, "html"))
# Connect and send
if settings.smtp_use_ssl:
server = smtplib.SMTP_SSL(settings.smtp_host, settings.smtp_port)
else:
server = smtplib.SMTP(settings.smtp_host, settings.smtp_port)
if settings.smtp_use_tls:
server.starttls()
server.login(settings.smtp_username, settings.smtp_password)
server.sendmail(settings.from_email, to_email, msg.as_string())
server.quit()
def _send_sendgrid_test(self, settings: VendorEmailSettings, to_email: str) -> None:
"""Send test email via SendGrid."""
try:
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail
except ImportError:
raise ValidationError("SendGrid library not installed. Contact support.")
message = Mail(
from_email=(settings.from_email, settings.from_name),
to_emails=to_email,
subject="Wizamart Email Configuration Test",
html_content=f"""
Email Configuration Test
This is a test email from Wizamart.
Your SendGrid settings are configured correctly!
""",
)
sg = SendGridAPIClient(settings.sendgrid_api_key)
response = sg.send(message)
if response.status_code >= 400:
raise Exception(f"SendGrid error: {response.status_code}")
def _send_mailgun_test(self, settings: VendorEmailSettings, to_email: str) -> None:
"""Send test email via Mailgun."""
import requests
response = requests.post(
f"https://api.mailgun.net/v3/{settings.mailgun_domain}/messages",
auth=("api", settings.mailgun_api_key),
data={
"from": f"{settings.from_name} <{settings.from_email}>",
"to": to_email,
"subject": "Wizamart Email Configuration Test",
"html": f"""
Email Configuration Test
This is a test email from Wizamart.
Your Mailgun settings are configured correctly!
""",
},
timeout=30,
)
if response.status_code >= 400:
raise Exception(f"Mailgun error: {response.text}")
def _send_ses_test(self, settings: VendorEmailSettings, to_email: str) -> None:
"""Send test email via Amazon SES."""
try:
import boto3
except ImportError:
raise ValidationError("boto3 library not installed. Contact support.")
client = boto3.client(
"ses",
region_name=settings.ses_region,
aws_access_key_id=settings.ses_access_key_id,
aws_secret_access_key=settings.ses_secret_access_key,
)
client.send_email(
Source=f"{settings.from_name} <{settings.from_email}>",
Destination={"ToAddresses": [to_email]},
Message={
"Subject": {"Data": "Wizamart Email Configuration Test"},
"Body": {
"Html": {
"Data": f"""
Email Configuration Test
This is a test email from Wizamart.
Your Amazon SES settings are configured correctly!
"""
}
},
},
)
# =========================================================================
# TIER HELPERS
# =========================================================================
def get_available_providers(self, tier: TierCode | None) -> list[dict]:
"""
Get list of available email providers for a tier.
Returns list of providers with availability status.
"""
providers = [
{
"code": EmailProvider.SMTP.value,
"name": "SMTP",
"description": "Standard SMTP email server",
"available": True,
"tier_required": None,
},
{
"code": EmailProvider.SENDGRID.value,
"name": "SendGrid",
"description": "SendGrid email delivery platform",
"available": tier in PREMIUM_TIERS if tier else False,
"tier_required": "business",
},
{
"code": EmailProvider.MAILGUN.value,
"name": "Mailgun",
"description": "Mailgun email API",
"available": tier in PREMIUM_TIERS if tier else False,
"tier_required": "business",
},
{
"code": EmailProvider.SES.value,
"name": "Amazon SES",
"description": "Amazon Simple Email Service",
"available": tier in PREMIUM_TIERS if tier else False,
"tier_required": "business",
},
]
return providers
# Module-level service factory
def get_vendor_email_settings_service(db: Session) -> VendorEmailSettingsService:
"""Factory function to get a VendorEmailSettingsService instance."""
return VendorEmailSettingsService(db)