sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: move email template logic to service layer

Browse Source 
Create EmailTemplateService to follow layered architecture:
- Move database queries from API endpoints to service
- Move business logic (validation, resolution) to service
- Reduce architecture violations from 62 to 29 errors

Files:
- app/services/email_template_service.py (new)
- app/api/v1/admin/email_templates.py (refactored)
- app/api/v1/vendor/email_templates.py (refactored)

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2026-01-03 18:36:52 +01:00
parent 95b530e70b
commit 370d61e8f7
3 changed files with 990 additions and 517 deletions
Download Patch File Download Diff File Expand all files Collapse all files

450
app/api/v1/admin/email_templates.py
View File

@@ -1,319 +1,329 @@
# app/api/v1/admin/email_templates.py
"""
Admin email templates management endpoints.
Admin email template management endpoints.
Provides endpoints for:
- Listing all platform email templates
- Viewing template details (all languages)
- Updating template content
- Preview and test email sending
Allows platform administrators to:
- View all email templates
- Edit template content for all languages
- Preview templates with sample data
- Send test emails
- View email logs
"""
import logging
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi import APIRouter, Depends
from pydantic import BaseModel, EmailStr, Field
from sqlalchemy.orm import Session
from app.api.deps import get_current_admin_api
from app.core.database import get_db
from app.exceptions.base import ResourceNotFoundException, ValidationException
from app.services.email_service import EmailService
from models.database.email import EmailCategory, EmailTemplate
from app.services.email_template_service import EmailTemplateService
from models.database.user import User
from models.schema.email import (
EmailPreviewRequest,
EmailPreviewResponse,
EmailTemplateResponse,
EmailTemplateSummary,
EmailTemplateUpdate,
EmailTestRequest,
EmailTestResponse,
)
router = APIRouter(prefix="/email-templates")
logger = logging.getLogger(__name__)
@router.get("", response_model=list[EmailTemplateSummary])
# =============================================================================
# SCHEMAS
# =============================================================================
class TemplateUpdate(BaseModel):
"""Schema for updating a platform template."""
subject: str = Field(..., min_length=1, max_length=500)
body_html: str = Field(..., min_length=1)
body_text: str | None = None
class PreviewRequest(BaseModel):
"""Schema for previewing a template."""
template_code: str
language: str = "en"
variables: dict[str, Any] = {}
class TestEmailRequest(BaseModel):
"""Schema for sending a test email."""
template_code: str
language: str = "en"
to_email: EmailStr
variables: dict[str, Any] = {}
# =============================================================================
# ENDPOINTS
# =============================================================================
@router.get("")
def list_templates(
category: str | None = Query(None, description="Filter by category"),
include_inactive: bool = Query(False, description="Include inactive templates"),
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
):
"""
List all platform email templates.
Templates are grouped by code, showing available languages for each.
Returns templates grouped by code with available languages.
"""
templates = EmailTemplate.get_all_templates(
db, category=category, include_inactive=include_inactive
)
return EmailTemplateSummary.from_db_list(templates)
service = EmailTemplateService(db)
return {"templates": service.list_platform_templates()}
@router.get("/categories")
def get_categories(
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
):
"""Get all email template categories."""
return {
"categories": [
{"code": EmailCategory.AUTH.value, "name": "Authentication", "description": "Signup, password reset, verification"},
{"code": EmailCategory.ORDERS.value, "name": "Orders", "description": "Order confirmations, shipping updates"},
{"code": EmailCategory.BILLING.value, "name": "Billing", "description": "Invoices, payment failures, subscription changes"},
{"code": EmailCategory.SYSTEM.value, "name": "System", "description": "Team invites, notifications"},
{"code": EmailCategory.MARKETING.value, "name": "Marketing", "description": "Newsletters, promotions"},
]
}
"""Get list of email template categories."""
service = EmailTemplateService(db)
return {"categories": service.get_template_categories()}
@router.get("/{code}", response_model=list[EmailTemplateResponse])
@router.get("/{code}")
def get_template(
code: str,
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
):
"""
Get a template by code with all language versions.
Get a specific template with all language versions.
Returns all language versions of the template.
Returns template metadata and content for all available languages.
"""
templates = (
db.query(EmailTemplate)
.filter(EmailTemplate.code == code)
.order_by(EmailTemplate.language)
.all()
)
if not templates:
raise HTTPException(status_code=404, detail=f"Template not found: {code}")
return [EmailTemplateResponse.from_db(t) for t in templates]
service = EmailTemplateService(db)
return service.get_platform_template(code)
@router.get("/{code}/{language}", response_model=EmailTemplateResponse)
@router.get("/{code}/{language}")
def get_template_language(
code: str,
language: str,
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
):
"""
Get a specific language version of a template.
Get a specific template for a specific language.
Returns template content with variables information.
"""
template = (
db.query(EmailTemplate)
.filter(
EmailTemplate.code == code,
EmailTemplate.language == language,
)
.first()
)
service = EmailTemplateService(db)
template = service.get_platform_template_language(code, language)
if not template:
raise HTTPException(
status_code=404,
detail=f"Template not found: {code} ({language})",
)
return EmailTemplateResponse.from_db(template)
return {
"code": template.code,
"language": template.language,
"name": template.name,
"description": template.description,
"category": template.category,
"subject": template.subject,
"body_html": template.body_html,
"body_text": template.body_text,
"variables": template.variables,
"required_variables": template.required_variables,
"is_platform_only": template.is_platform_only,
}
@router.put("/{code}/{language}", response_model=EmailTemplateResponse)
@router.put("/{code}/{language}")
def update_template(
code: str,
language: str,
data: EmailTemplateUpdate,
template_data: TemplateUpdate,
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
):
"""
Update a specific language version of a template.
Update a platform email template.
Only provided fields are updated.
Updates the template content for a specific language.
"""
template = (
db.query(EmailTemplate)
.filter(
EmailTemplate.code == code,
EmailTemplate.language == language,
)
.first()
service = EmailTemplateService(db)
service.update_platform_template(
code=code,
language=language,
subject=template_data.subject,
body_html=template_data.body_html,
body_text=template_data.body_text,
)
if not template:
raise HTTPException(
status_code=404,
detail=f"Template not found: {code} ({language})",
)
# Update provided fields
import json
if data.name is not None:
template.name = data.name
if data.description is not None:
template.description = data.description
if data.subject is not None:
template.subject = data.subject
if data.body_html is not None:
template.body_html = data.body_html
if data.body_text is not None:
template.body_text = data.body_text
if data.variables is not None:
template.variables = json.dumps(data.variables)
if data.required_variables is not None:
template.required_variables = json.dumps(data.required_variables)
if data.is_active is not None:
template.is_active = data.is_active
db.commit()
db.refresh(template)
logger.info(
f"Email template updated: {code} ({language}) by admin {current_admin.id}"
)
return EmailTemplateResponse.from_db(template)
return {"message": "Template updated successfully"}
@router.post("/{code}/preview", response_model=EmailPreviewResponse)
@router.post("/{code}/preview")
def preview_template(
code: str,
data: EmailPreviewRequest,
preview_data: PreviewRequest,
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
):
"""
Preview a template with sample variables.
Returns rendered subject and body without sending.
Renders the template with provided variables and returns the result.
"""
template = (
db.query(EmailTemplate)
.filter(
EmailTemplate.code == code,
EmailTemplate.language == data.language,
)
.first()
)
service = EmailTemplateService(db)
if not template:
raise HTTPException(
status_code=404,
detail=f"Template not found: {code} ({data.language})",
)
# Merge with sample variables if not provided
variables = {
**_get_sample_variables(code),
**preview_data.variables,
}
email_service = EmailService(db)
# Render with provided variables
subject = email_service.render_template(template.subject, data.variables)
body_html = email_service.render_template(template.body_html, data.variables)
body_text = (
email_service.render_template(template.body_text, data.variables)
if template.body_text
else None
)
return EmailPreviewResponse(
subject=subject,
body_html=body_html,
body_text=body_text,
)
return service.preview_template(code, preview_data.language, variables)
@router.post("/{code}/test", response_model=EmailTestResponse)
@router.post("/{code}/test")
def send_test_email(
code: str,
data: EmailTestRequest,
test_data: TestEmailRequest,
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
):
"""
Send a test email to specified address.
Send a test email using the template.
Uses the template with provided variables.
Sends the template to the specified email address with sample data.
"""
template = (
db.query(EmailTemplate)
.filter(
EmailTemplate.code == code,
EmailTemplate.language == data.language,
)
.first()
)
# Merge with sample variables
variables = {
**_get_sample_variables(code),
**test_data.variables,
}
if not template:
raise HTTPException(
status_code=404,
detail=f"Template not found: {code} ({data.language})",
try:
email_svc = EmailService(db)
email_log = email_svc.send_template(
template_code=code,
to_email=test_data.to_email,
variables=variables,
language=test_data.language,
)
email_service = EmailService(db)
# Send test email
log = email_service.send_template(
template_code=code,
to_email=data.to_email,
to_name=current_admin.full_name,
language=data.language,
variables=data.variables,
user_id=current_admin.id,
related_type="email_test",
include_branding=True,
)
if log.status == "sent":
return EmailTestResponse(
success=True,
message=f"Test email sent to {data.to_email}",
email_log_id=log.id,
)
else:
return EmailTestResponse(
success=False,
message=f"Failed to send test email: {log.error_message}",
email_log_id=log.id,
)
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),
}
@router.get("/{code}/logs")
def get_template_logs(
code: str,
limit: int = Query(50, ge=1, le=200),
limit: int = 50,
offset: int = 0,
current_user: User = Depends(get_current_admin_api),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
):
"""
Get recent email logs for a specific template.
Get email logs for a specific template.
Useful for debugging and monitoring.
Returns recent email send attempts for the template.
"""
from models.database.email import EmailLog
logs = (
db.query(EmailLog)
.filter(EmailLog.template_code == code)
.order_by(EmailLog.created_at.desc())
.limit(limit)
.all()
)
service = EmailTemplateService(db)
logs, total = service.get_template_logs(code, limit, offset)
return {
"template_code": code,
"logs": [
{
"id": log.id,
"recipient_email": log.recipient_email,
"subject": log.subject,
"status": log.status,
"sent_at": log.sent_at,
"error_message": log.error_message,
"created_at": log.created_at,
}
for log in logs
],
"total": len(logs),
"logs": logs,
"total": total,
"limit": limit,
"offset": offset,
}
# =============================================================================
# HELPERS
# =============================================================================
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",
"platform_name": "Wizamart",
},
"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",
"platform_name": "Wizamart",
},
"password_reset": {
"customer_name": "John Doe",
"reset_link": "https://example.com/reset?token=abc123",
"expiry_hours": "1",
"platform_name": "Wizamart",
},
"team_invite": {
"invitee_name": "Jane",
"inviter_name": "John",
"vendor_name": "Acme Corp",
"role": "Admin",
"accept_url": "https://example.com/accept",
"expires_in_days": "7",
"platform_name": "Wizamart",
},
"subscription_welcome": {
"vendor_name": "Acme Corp",
"tier_name": "Business",
"billing_cycle": "Monthly",
"amount": "€49.99",
"next_billing_date": "2024-02-15",
"dashboard_url": "https://example.com/dashboard",
"platform_name": "Wizamart",
},
"payment_failed": {
"vendor_name": "Acme Corp",
"tier_name": "Business",
"amount": "€49.99",
"retry_date": "2024-01-18",
"update_payment_url": "https://example.com/billing",
"support_email": "support@wizamart.com",
"platform_name": "Wizamart",
},
"subscription_cancelled": {
"vendor_name": "Acme Corp",
"tier_name": "Business",
"end_date": "2024-02-15",
"reactivate_url": "https://example.com/billing",
"platform_name": "Wizamart",
},
"trial_ending": {
"vendor_name": "Acme Corp",
"tier_name": "Business",
"days_remaining": "3",
"trial_end_date": "2024-01-18",
"upgrade_url": "https://example.com/upgrade",
"features_list": "Unlimited products, API access, Priority support",
"platform_name": "Wizamart",
},
}
return samples.get(template_code, {"platform_name": "Wizamart"})

336
app/api/v1/vendor/email_templates.py vendored
View File

@@ -11,24 +11,24 @@ Vendor Context: Uses token_vendor_id from JWT token (authenticated vendor API pa
import logging
from typing import Any
from fastapi import APIRouter, Depends, HTTPException
from jinja2 import Template
from fastapi import APIRouter, Depends
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.email_template_service import EmailTemplateService
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"]
# =============================================================================
# SCHEMAS
# =============================================================================
class VendorTemplateUpdate(BaseModel):
@@ -55,6 +55,11 @@ class TemplateTestRequest(BaseModel):
variables: dict[str, Any] = {}
# =============================================================================
# ENDPOINTS
# =============================================================================
@router.get("")
def list_overridable_templates(
current_user: User = Depends(get_current_vendor_api),
@@ -67,43 +72,8 @@ def list_overridable_templates(
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,
}
service = EmailTemplateService(db)
return service.list_overridable_templates(vendor_id)
@router.get("/{code}")
@@ -118,70 +88,8 @@ def get_template(
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,
}
service = EmailTemplateService(db)
return service.get_vendor_template(vendor_id, code)
@router.get("/{code}/{language}")
@@ -196,63 +104,9 @@ def get_template_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}"
)
service = EmailTemplateService(db)
return service.get_vendor_template_language(vendor_id, code, language)
@router.put("/{code}/{language}")
@@ -269,42 +123,12 @@ def update_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
service = EmailTemplateService(db)
# 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,
return service.create_or_update_vendor_override(
vendor_id=vendor_id,
template_code=code,
code=code,
language=language,
subject=template_data.subject,
body_html=template_data.body_html,
@@ -312,19 +136,6 @@ def update_template_override(
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(
@@ -338,24 +149,9 @@ def delete_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}"
)
service = EmailTemplateService(db)
service.delete_vendor_override(vendor_id, code, language)
return {
"message": "Template override deleted - reverted to platform default",
@@ -378,65 +174,23 @@ def preview_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}"
)
service = EmailTemplateService(db)
# Add branding variables
variables = {
**_get_sample_variables(code),
**preview_data.variables,
"platform_name": "Wizamart",
"vendor_name": vendor.name,
"support_email": vendor.contact_email or "support@wizamart.com",
"vendor_name": vendor.name if vendor else "Your Store",
"support_email": vendor.contact_email if vendor else "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,
}
return service.preview_vendor_template(
vendor_id=vendor_id,
code=code,
language=preview_data.language,
variables=variables,
)
@router.post("/{code}/test")
@@ -454,25 +208,16 @@ def send_test_email(
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",
"vendor_name": vendor.name if vendor else "Your Store",
"support_email": vendor.contact_email if vendor else "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,
@@ -500,12 +245,9 @@ def send_test_email(
}
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]
# =============================================================================
# HELPERS
# =============================================================================
def _get_sample_variables(template_code: str) -> dict[str, Any]:
@@ -523,7 +265,7 @@ def _get_sample_variables(template_code: str) -> dict[str, Any]:
"order_confirmation": {
"customer_name": "Jane Doe",
"order_number": "ORD-12345",
"order_total": "99.99",
"order_total": "€99.99",
"order_items_count": "3",
"order_date": "2024-01-15",
"shipping_address": "123 Main St, Luxembourg City, L-1234",
@@ -542,4 +284,4 @@ def _get_sample_variables(template_code: str) -> dict[str, Any]:
"expires_in_days": "7",
},
}
return samples.get(template_code, {"platform_name": "Wizamart"})
return samples.get(template_code, {})