Files
orion/app/api/v1/vendor/email_templates.py
Samir Boulahtit c52af2a155 feat: implement email template system with vendor overrides
Add comprehensive email template management for both admin and vendors:

Admin Features:
- Email templates management page at /admin/email-templates
- Edit platform templates with language support (en, fr, de, lb)
- Preview templates with sample variables
- Send test emails
- View email logs per template

Vendor Features:
- Email templates customization page at /vendor/{code}/email-templates
- Override platform templates with vendor-specific versions
- Preview and test customized templates
- Revert to platform defaults

Technical Changes:
- Migration for vendor_email_templates table
- VendorEmailTemplate model with override management
- Enhanced EmailService with language resolution chain
  (customer preferred -> vendor preferred -> platform default)
- Branding resolution (Wizamart default, removed for whitelabel)
- Platform-only template protection (billing templates)
- Admin and vendor API endpoints with full CRUD
- Updated seed script with billing and team templates

Files: 22 changed, ~3,900 lines added

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-03 18:29:26 +01:00

546 lines
17 KiB
Python

# app/api/v1/vendor/email_templates.py
"""
Vendor email template override endpoints.
Allows vendors to customize platform email templates with their own content.
Platform-only templates (billing, subscription) cannot be overridden.
Vendor Context: Uses token_vendor_id from JWT token (authenticated vendor API pattern).
"""
import logging
from typing import Any
from fastapi import APIRouter, Depends, HTTPException
from jinja2 import Template
from pydantic import BaseModel, EmailStr, Field
from sqlalchemy.orm import Session
from app.api.deps import get_current_vendor_api
from app.core.database import get_db
from app.services.email_service import EmailService
from app.services.vendor_service import vendor_service
from models.database.email import EmailTemplate
from models.database.user import User
from models.database.vendor_email_template import VendorEmailTemplate
router = APIRouter(prefix="/email-templates")
logger = logging.getLogger(__name__)
# Supported languages
SUPPORTED_LANGUAGES = ["en", "fr", "de", "lb"]
class VendorTemplateUpdate(BaseModel):
"""Schema for creating/updating a vendor template override."""
subject: str = Field(..., min_length=1, max_length=500)
body_html: str = Field(..., min_length=1)
body_text: str | None = None
name: str | None = Field(None, max_length=255)
class TemplatePreviewRequest(BaseModel):
"""Schema for previewing a template."""
language: str = "en"
variables: dict[str, Any] = {}
class TemplateTestRequest(BaseModel):
"""Schema for sending a test email."""
to_email: EmailStr
language: str = "en"
variables: dict[str, Any] = {}
@router.get("")
def list_overridable_templates(
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
):
"""
List all email templates that the vendor can customize.
Returns platform templates with vendor override status.
Platform-only templates (billing, subscription) are excluded.
"""
vendor_id = current_user.token_vendor_id
# Get all overridable platform templates
platform_templates = EmailTemplate.get_overridable_templates(db)
# Get all vendor overrides
vendor_overrides = VendorEmailTemplate.get_all_overrides_for_vendor(db, vendor_id)
# Build override lookup
override_lookup = {}
for override in vendor_overrides:
key = (override.template_code, override.language)
override_lookup[key] = override
# Build response
templates_response = []
for template in platform_templates:
# Check which languages have overrides
override_languages = []
for lang in SUPPORTED_LANGUAGES:
if (template.code, lang) in override_lookup:
override_languages.append(lang)
templates_response.append({
"code": template.code,
"name": template.name,
"category": template.category,
"description": template.description,
"available_languages": _get_available_languages(db, template.code),
"override_languages": override_languages,
"has_override": len(override_languages) > 0,
"variables": template.required_variables.split(",") if template.required_variables else [],
})
return {
"templates": templates_response,
"supported_languages": SUPPORTED_LANGUAGES,
}
@router.get("/{code}")
def get_template(
code: str,
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
):
"""
Get a specific template with all language versions.
Returns platform template details and vendor overrides for each language.
"""
vendor_id = current_user.token_vendor_id
# Get platform template
platform_template = db.query(EmailTemplate).filter(
EmailTemplate.code == code
).first()
if not platform_template:
raise HTTPException(status_code=404, detail="Template not found")
if platform_template.is_platform_only:
raise HTTPException(
status_code=403,
detail="This is a platform-only template and cannot be customized"
)
# Get all language versions of platform template
platform_versions = db.query(EmailTemplate).filter(
EmailTemplate.code == code
).all()
# Get vendor overrides for all languages
vendor_overrides = (
db.query(VendorEmailTemplate)
.filter(
VendorEmailTemplate.vendor_id == vendor_id,
VendorEmailTemplate.template_code == code,
)
.all()
)
override_lookup = {v.language: v for v in vendor_overrides}
platform_lookup = {t.language: t for t in platform_versions}
# Build language versions
languages = {}
for lang in SUPPORTED_LANGUAGES:
platform_ver = platform_lookup.get(lang)
override_ver = override_lookup.get(lang)
languages[lang] = {
"has_platform_template": platform_ver is not None,
"has_vendor_override": override_ver is not None,
"platform": {
"subject": platform_ver.subject if platform_ver else None,
"body_html": platform_ver.body_html if platform_ver else None,
"body_text": platform_ver.body_text if platform_ver else None,
} if platform_ver else None,
"vendor_override": {
"subject": override_ver.subject if override_ver else None,
"body_html": override_ver.body_html if override_ver else None,
"body_text": override_ver.body_text if override_ver else None,
"name": override_ver.name if override_ver else None,
"updated_at": override_ver.updated_at.isoformat() if override_ver else None,
} if override_ver else None,
}
return {
"code": code,
"name": platform_template.name,
"category": platform_template.category,
"description": platform_template.description,
"variables": platform_template.required_variables.split(",") if platform_template.required_variables else [],
"languages": languages,
}
@router.get("/{code}/{language}")
def get_template_language(
code: str,
language: str,
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
):
"""
Get a specific template for a specific language.
Returns vendor override if exists, otherwise platform template.
"""
if language not in SUPPORTED_LANGUAGES:
raise HTTPException(status_code=400, detail=f"Unsupported language: {language}")
vendor_id = current_user.token_vendor_id
# Check if template is overridable
platform_template = db.query(EmailTemplate).filter(
EmailTemplate.code == code
).first()
if not platform_template:
raise HTTPException(status_code=404, detail="Template not found")
if platform_template.is_platform_only:
raise HTTPException(
status_code=403,
detail="This is a platform-only template and cannot be customized"
)
# Check for vendor override
vendor_override = VendorEmailTemplate.get_override(db, vendor_id, code, language)
# Get platform version for this language
platform_version = EmailTemplate.get_by_code_and_language(db, code, language)
if vendor_override:
return {
"code": code,
"language": language,
"source": "vendor_override",
"subject": vendor_override.subject,
"body_html": vendor_override.body_html,
"body_text": vendor_override.body_text,
"name": vendor_override.name,
"variables": platform_template.required_variables.split(",") if platform_template.required_variables else [],
"platform_template": {
"subject": platform_version.subject if platform_version else None,
"body_html": platform_version.body_html if platform_version else None,
} if platform_version else None,
}
elif platform_version:
return {
"code": code,
"language": language,
"source": "platform",
"subject": platform_version.subject,
"body_html": platform_version.body_html,
"body_text": platform_version.body_text,
"name": platform_version.name,
"variables": platform_template.required_variables.split(",") if platform_template.required_variables else [],
"platform_template": None,
}
else:
raise HTTPException(
status_code=404,
detail=f"No template found for language: {language}"
)
@router.put("/{code}/{language}")
def update_template_override(
code: str,
language: str,
template_data: VendorTemplateUpdate,
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
):
"""
Create or update a vendor template override.
Creates a vendor-specific version of the email template.
The platform template remains unchanged.
"""
if language not in SUPPORTED_LANGUAGES:
raise HTTPException(status_code=400, detail=f"Unsupported language: {language}")
vendor_id = current_user.token_vendor_id
# Check if template exists and is overridable
platform_template = db.query(EmailTemplate).filter(
EmailTemplate.code == code
).first()
if not platform_template:
raise HTTPException(status_code=404, detail="Template not found")
if platform_template.is_platform_only:
raise HTTPException(
status_code=403,
detail="This is a platform-only template and cannot be customized"
)
# Validate template content (try to render with dummy variables)
try:
Template(template_data.subject).render({})
Template(template_data.body_html).render({})
if template_data.body_text:
Template(template_data.body_text).render({})
except Exception as e:
raise HTTPException(
status_code=400,
detail=f"Invalid template syntax: {str(e)}"
)
# Create or update override
override = VendorEmailTemplate.create_or_update(
db=db,
vendor_id=vendor_id,
template_code=code,
language=language,
subject=template_data.subject,
body_html=template_data.body_html,
body_text=template_data.body_text,
name=template_data.name,
)
db.commit()
logger.info(
f"Vendor {vendor_id} updated email template override: {code}/{language}"
)
return {
"message": "Template override saved",
"code": code,
"language": language,
"is_new": override.created_at == override.updated_at,
}
@router.delete("/{code}/{language}")
def delete_template_override(
code: str,
language: str,
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
):
"""
Delete a vendor template override.
Reverts to using the platform default template for this language.
"""
if language not in SUPPORTED_LANGUAGES:
raise HTTPException(status_code=400, detail=f"Unsupported language: {language}")
vendor_id = current_user.token_vendor_id
deleted = VendorEmailTemplate.delete_override(db, vendor_id, code, language)
if not deleted:
raise HTTPException(
status_code=404,
detail="No override found for this template and language"
)
db.commit()
logger.info(
f"Vendor {vendor_id} deleted email template override: {code}/{language}"
)
return {
"message": "Template override deleted - reverted to platform default",
"code": code,
"language": language,
}
@router.post("/{code}/preview")
def preview_template(
code: str,
preview_data: TemplatePreviewRequest,
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
):
"""
Preview a template with sample variables.
Uses vendor override if exists, otherwise platform template.
"""
vendor_id = current_user.token_vendor_id
vendor = vendor_service.get_vendor_by_id(db, vendor_id)
# Check if template exists
platform_template = db.query(EmailTemplate).filter(
EmailTemplate.code == code
).first()
if not platform_template:
raise HTTPException(status_code=404, detail="Template not found")
# Get template content (vendor override or platform)
vendor_override = VendorEmailTemplate.get_override(
db, vendor_id, code, preview_data.language
)
platform_version = EmailTemplate.get_by_code_and_language(
db, code, preview_data.language
)
if vendor_override:
subject = vendor_override.subject
body_html = vendor_override.body_html
body_text = vendor_override.body_text
source = "vendor_override"
elif platform_version:
subject = platform_version.subject
body_html = platform_version.body_html
body_text = platform_version.body_text
source = "platform"
else:
raise HTTPException(
status_code=404,
detail=f"No template found for language: {preview_data.language}"
)
# Add branding variables
variables = {
**preview_data.variables,
"platform_name": "Wizamart",
"vendor_name": vendor.name,
"support_email": vendor.contact_email or "support@wizamart.com",
}
# Render templates
try:
rendered_subject = Template(subject).render(variables)
rendered_html = Template(body_html).render(variables)
rendered_text = Template(body_text).render(variables) if body_text else None
except Exception as e:
raise HTTPException(
status_code=400,
detail=f"Template rendering error: {str(e)}"
)
return {
"source": source,
"language": preview_data.language,
"subject": rendered_subject,
"body_html": rendered_html,
"body_text": rendered_text,
}
@router.post("/{code}/test")
def send_test_email(
code: str,
test_data: TemplateTestRequest,
current_user: User = Depends(get_current_vendor_api),
db: Session = Depends(get_db),
):
"""
Send a test email using the template.
Uses vendor override if exists, otherwise platform template.
"""
vendor_id = current_user.token_vendor_id
vendor = vendor_service.get_vendor_by_id(db, vendor_id)
# Check if template exists
platform_template = db.query(EmailTemplate).filter(
EmailTemplate.code == code
).first()
if not platform_template:
raise HTTPException(status_code=404, detail="Template not found")
# Build test variables
variables = {
**_get_sample_variables(code),
**test_data.variables,
"platform_name": "Wizamart",
"vendor_name": vendor.name,
"support_email": vendor.contact_email or "support@wizamart.com",
}
try:
# Send using email service (will use vendor override if exists)
email_svc = EmailService(db)
email_log = email_svc.send_template(
template_code=code,
to_email=test_data.to_email,
variables=variables,
vendor_id=vendor_id,
language=test_data.language,
)
if email_log.status == "sent":
return {
"success": True,
"message": f"Test email sent to {test_data.to_email}",
}
else:
return {
"success": False,
"message": email_log.error_message or "Failed to send email",
}
except Exception as e:
logger.exception(f"Failed to send test email: {e}")
return {
"success": False,
"message": str(e),
}
def _get_available_languages(db: Session, code: str) -> list[str]:
"""Get list of languages that have platform templates."""
templates = db.query(EmailTemplate.language).filter(
EmailTemplate.code == code
).all()
return [t.language for t in templates]
def _get_sample_variables(template_code: str) -> dict[str, Any]:
"""Get sample variables for testing templates."""
samples = {
"signup_welcome": {
"first_name": "John",
"company_name": "Acme Corp",
"email": "john@example.com",
"vendor_code": "acme",
"login_url": "https://example.com/login",
"trial_days": "14",
"tier_name": "Business",
},
"order_confirmation": {
"customer_name": "Jane Doe",
"order_number": "ORD-12345",
"order_total": "99.99",
"order_items_count": "3",
"order_date": "2024-01-15",
"shipping_address": "123 Main St, Luxembourg City, L-1234",
},
"password_reset": {
"customer_name": "John Doe",
"reset_link": "https://example.com/reset?token=abc123",
"expiry_hours": "1",
},
"team_invite": {
"invitee_name": "Jane",
"inviter_name": "John",
"vendor_name": "Acme Corp",
"role": "Admin",
"accept_url": "https://example.com/accept",
"expires_in_days": "7",
},
}
return samples.get(template_code, {"platform_name": "Wizamart"})