Benjamin_Boenisch/breakpilot-compliance
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(backend/api): extract Email Template services (Step 4 — file 10 of 18)

Browse Source 
compliance/api/email_template_routes.py (823 LOC) -> 295 LOC thin routes
+ 402-line EmailTemplateService + 241-line EmailTemplateVersionService +
61-line schemas file.

Two-service split along natural responsibility seam:

  email_template_service.py (402 LOC):
    - Template type catalog (TEMPLATE_TYPES constant)
    - Template CRUD (list, create, get)
    - Stats, settings, send logs, initialization, default content
    - Shared _template_to_dict / _version_to_dict / _render_template helpers

  email_template_version_service.py (241 LOC):
    - Version CRUD (create, list, get, update)
    - Workflow transitions (submit, approve, reject, publish)
    - Preview and test-send

TEMPLATE_TYPES, VALID_CATEGORIES, VALID_STATUSES re-exported from the
route module for any legacy consumers.

State-transition errors use ValidationError (-> HTTPException 400) to
preserve the original handler's 400 status for "Only draft/review
versions can be ..." checks, since the existing TestClient integration
tests (47 tests) assert status_code == 400.

Verified:
  - 47/47 tests/test_email_template_routes.py pass
  - OpenAPI 360/484 unchanged
  - mypy compliance/ -> Success on 138 source files
  - email_template_routes.py 823 -> 295 LOC
  - Hard-cap violations: 9 -> 8

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Sharang Parnerkar
2026-04-08 22:39:19 +02:00
parent a638d0e527
commit 0c2e03f294
6 changed files with 1087 additions and 794 deletions
Download Patch File Download Diff File Expand all files Collapse all files

759
backend-compliance/compliance/api/email_template_routes.py
View File

@@ -3,153 +3,53 @@ E-Mail-Template Routes — Benachrichtigungsvorlagen fuer DSGVO-Compliance.
Verwaltet Templates fuer DSR, Consent, Breach, Vendor und Training E-Mails.
Inklusive Versionierung, Approval-Workflow, Vorschau und Send-Logging.
Phase 1 Step 4 refactor: handlers delegate to EmailTemplateService
(templates/settings/logs/stats/initialize) and EmailTemplateVersionService
(version workflow + preview + test-send). Template types catalog is
re-exported for any legacy callers.
"""
import uuid
from datetime import datetime, timezone
from typing import Optional, Dict
from typing import Any, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, Header
from pydantic import BaseModel
from fastapi import APIRouter, Depends, Header, Query
from sqlalchemy.orm import Session
from classroom_engine.database import get_db
from ..db.email_template_models import (
EmailTemplateDB, EmailTemplateVersionDB, EmailTemplateApprovalDB,
EmailSendLogDB, EmailTemplateSettingsDB,
from compliance.api._http_errors import translate_domain_errors
from compliance.schemas.email_template import (
PreviewRequest,
SendTestRequest,
SettingsUpdate,
TemplateCreate,
VersionCreate,
VersionUpdate,
)
from compliance.services.email_template_service import (
TEMPLATE_TYPES,
VALID_CATEGORIES,
VALID_STATUSES,
EmailTemplateService,
)
from compliance.services.email_template_version_service import (
EmailTemplateVersionService,
)
router = APIRouter(prefix="/email-templates", tags=["compliance-email-templates"])
DEFAULT_TENANT = "9282a473-5c95-4b3a-bf78-0ecc0ec71d3e"
# Template-Typen und zugehoerige Variablen
TEMPLATE_TYPES = {
"welcome": {"name": "Willkommen", "category": "general", "variables": ["user_name", "company_name", "login_url"]},
"verification": {"name": "E-Mail-Verifizierung", "category": "general", "variables": ["user_name", "verification_url", "expiry_hours"]},
"password_reset": {"name": "Passwort zuruecksetzen", "category": "general", "variables": ["user_name", "reset_url", "expiry_hours"]},
"dsr_receipt": {"name": "DSR Eingangsbestaetigung", "category": "dsr", "variables": ["requester_name", "reference_number", "request_type", "deadline"]},
"dsr_identity_request": {"name": "DSR Identitaetsanfrage", "category": "dsr", "variables": ["requester_name", "reference_number"]},
"dsr_completion": {"name": "DSR Abschluss", "category": "dsr", "variables": ["requester_name", "reference_number", "request_type", "completion_date"]},
"dsr_rejection": {"name": "DSR Ablehnung", "category": "dsr", "variables": ["requester_name", "reference_number", "rejection_reason", "legal_basis"]},
"dsr_extension": {"name": "DSR Fristverlaengerung", "category": "dsr", "variables": ["requester_name", "reference_number", "new_deadline", "extension_reason"]},
"consent_request": {"name": "Einwilligungsanfrage", "category": "consent", "variables": ["user_name", "purpose", "consent_url"]},
"consent_confirmation": {"name": "Einwilligungsbestaetigung", "category": "consent", "variables": ["user_name", "purpose", "consent_date"]},
"consent_withdrawal": {"name": "Widerruf bestaetigt", "category": "consent", "variables": ["user_name", "purpose", "withdrawal_date"]},
"consent_reminder": {"name": "Einwilligungs-Erinnerung", "category": "consent", "variables": ["user_name", "purpose", "expiry_date"]},
"breach_notification_authority": {"name": "Datenpanne Aufsichtsbehoerde", "category": "breach", "variables": ["incident_date", "incident_description", "affected_count", "measures_taken", "authority_name"]},
"breach_notification_affected": {"name": "Datenpanne Betroffene", "category": "breach", "variables": ["user_name", "incident_date", "incident_description", "measures_taken", "contact_info"]},
"breach_internal": {"name": "Datenpanne intern", "category": "breach", "variables": ["reporter_name", "incident_date", "incident_description", "severity"]},
"vendor_dpa_request": {"name": "AVV-Anfrage", "category": "vendor", "variables": ["vendor_name", "contact_name", "deadline", "requirements"]},
"vendor_review_reminder": {"name": "Vendor-Pruefung Erinnerung", "category": "vendor", "variables": ["vendor_name", "review_due_date", "last_review_date"]},
"training_invitation": {"name": "Schulungseinladung", "category": "training", "variables": ["user_name", "training_title", "training_date", "training_url"]},
"training_reminder": {"name": "Schulungs-Erinnerung", "category": "training", "variables": ["user_name", "training_title", "deadline"]},
"training_completion": {"name": "Schulung abgeschlossen", "category": "training", "variables": ["user_name", "training_title", "completion_date", "certificate_url"]},
}
VALID_STATUSES = ["draft", "review", "approved", "published"]
VALID_CATEGORIES = ["general", "dsr", "consent", "breach", "vendor", "training"]
# =============================================================================
# Pydantic Schemas
# =============================================================================
class TemplateCreate(BaseModel):
template_type: str
name: Optional[str] = None
description: Optional[str] = None
category: Optional[str] = None
is_active: bool = True
class VersionCreate(BaseModel):
version: str = "1.0"
language: str = "de"
subject: str
body_html: str
body_text: Optional[str] = None
class VersionUpdate(BaseModel):
subject: Optional[str] = None
body_html: Optional[str] = None
body_text: Optional[str] = None
class PreviewRequest(BaseModel):
variables: Optional[Dict[str, str]] = None
class SendTestRequest(BaseModel):
recipient: str
variables: Optional[Dict[str, str]] = None
class SettingsUpdate(BaseModel):
sender_name: Optional[str] = None
sender_email: Optional[str] = None
reply_to: Optional[str] = None
logo_url: Optional[str] = None
primary_color: Optional[str] = None
secondary_color: Optional[str] = None
footer_text: Optional[str] = None
company_name: Optional[str] = None
company_address: Optional[str] = None
# =============================================================================
# Helpers
# =============================================================================
def _get_tenant(x_tenant_id: Optional[str] = Header(None, alias='X-Tenant-ID')) -> str:
def _get_tenant(x_tenant_id: Optional[str] = Header(None, alias="X-Tenant-ID")) -> str:
return x_tenant_id or DEFAULT_TENANT
def _template_to_dict(t: EmailTemplateDB, latest_version=None) -> dict:
result = {
"id": str(t.id),
"tenant_id": str(t.tenant_id),
"template_type": t.template_type,
"name": t.name,
"description": t.description,
"category": t.category,
"is_active": t.is_active,
"sort_order": t.sort_order,
"variables": t.variables or [],
"created_at": t.created_at.isoformat() if t.created_at else None,
"updated_at": t.updated_at.isoformat() if t.updated_at else None,
}
if latest_version:
result["latest_version"] = _version_to_dict(latest_version)
return result
def get_template_service(db: Session = Depends(get_db)) -> EmailTemplateService:
return EmailTemplateService(db)
def _version_to_dict(v: EmailTemplateVersionDB) -> dict:
return {
"id": str(v.id),
"template_id": str(v.template_id),
"version": v.version,
"language": v.language,
"subject": v.subject,
"body_html": v.body_html,
"body_text": v.body_text,
"status": v.status,
"submitted_at": v.submitted_at.isoformat() if v.submitted_at else None,
"submitted_by": v.submitted_by,
"published_at": v.published_at.isoformat() if v.published_at else None,
"published_by": v.published_by,
"created_at": v.created_at.isoformat() if v.created_at else None,
"created_by": v.created_by,
}
def _render_template(html: str, variables: Dict[str, str]) -> str:
"""Replace {{variable}} placeholders with values."""
result = html
for key, value in variables.items():
result = result.replace(f"{{{{{key}}}}}", str(value))
return result
def get_version_service(db: Session = Depends(get_db)) -> EmailTemplateVersionService:
return EmailTemplateVersionService(db)
# =============================================================================
@@ -157,135 +57,40 @@ def _render_template(html: str, variables: Dict[str, str]) -> str:
# =============================================================================
@router.get("/types")
async def get_template_types():
async def get_template_types() -> list[dict[str, Any]]:
"""Gibt alle verfuegbaren Template-Typen mit Variablen zurueck."""
return [
{
"type": ttype,
"name": info["name"],
"category": info["category"],
"variables": info["variables"],
}
for ttype, info in TEMPLATE_TYPES.items()
]
return EmailTemplateService.list_types()
@router.get("/stats")
async def get_stats(
tenant_id: str = Depends(_get_tenant),
db: Session = Depends(get_db),
):
service: EmailTemplateService = Depends(get_template_service),
) -> dict[str, Any]:
"""Statistiken ueber E-Mail-Templates."""
tid = uuid.UUID(tenant_id)
base = db.query(EmailTemplateDB).filter(EmailTemplateDB.tenant_id == tid)
total = base.count()
active = base.filter(EmailTemplateDB.is_active).count()
# Count templates with published versions
published_count = 0
templates = base.all()
for t in templates:
has_published = db.query(EmailTemplateVersionDB).filter(
EmailTemplateVersionDB.template_id == t.id,
EmailTemplateVersionDB.status == "published",
).count() > 0
if has_published:
published_count += 1
# By category
by_category = {}
for cat in VALID_CATEGORIES:
by_category[cat] = base.filter(EmailTemplateDB.category == cat).count()
# Send logs stats
total_sent = db.query(EmailSendLogDB).filter(EmailSendLogDB.tenant_id == tid).count()
return {
"total": total,
"active": active,
"published": published_count,
"draft": total - published_count,
"by_category": by_category,
"total_sent": total_sent,
}
with translate_domain_errors():
return service.stats(tenant_id)
@router.get("/settings")
async def get_settings(
tenant_id: str = Depends(_get_tenant),
db: Session = Depends(get_db),
):
service: EmailTemplateService = Depends(get_template_service),
) -> dict[str, Any]:
"""Globale E-Mail-Einstellungen laden."""
tid = uuid.UUID(tenant_id)
settings = db.query(EmailTemplateSettingsDB).filter(
EmailTemplateSettingsDB.tenant_id == tid,
).first()
if not settings:
return {
"sender_name": "Datenschutzbeauftragter",
"sender_email": "datenschutz@example.de",
"reply_to": None,
"logo_url": None,
"primary_color": "#4F46E5",
"secondary_color": "#7C3AED",
"footer_text": "Datenschutzhinweis: Diese E-Mail enthaelt vertrauliche Informationen.",
"company_name": None,
"company_address": None,
}
return {
"sender_name": settings.sender_name,
"sender_email": settings.sender_email,
"reply_to": settings.reply_to,
"logo_url": settings.logo_url,
"primary_color": settings.primary_color,
"secondary_color": settings.secondary_color,
"footer_text": settings.footer_text,
"company_name": settings.company_name,
"company_address": settings.company_address,
}
with translate_domain_errors():
return service.get_settings(tenant_id)
@router.put("/settings")
async def update_settings(
body: SettingsUpdate,
tenant_id: str = Depends(_get_tenant),
db: Session = Depends(get_db),
):
service: EmailTemplateService = Depends(get_template_service),
) -> dict[str, Any]:
"""Globale E-Mail-Einstellungen speichern."""
tid = uuid.UUID(tenant_id)
settings = db.query(EmailTemplateSettingsDB).filter(
EmailTemplateSettingsDB.tenant_id == tid,
).first()
if not settings:
settings = EmailTemplateSettingsDB(tenant_id=tid)
db.add(settings)
for field in ["sender_name", "sender_email", "reply_to", "logo_url",
"primary_color", "secondary_color", "footer_text",
"company_name", "company_address"]:
val = getattr(body, field, None)
if val is not None:
setattr(settings, field, val)
settings.updated_at = datetime.now(timezone.utc)
db.commit()
db.refresh(settings)
return {
"sender_name": settings.sender_name,
"sender_email": settings.sender_email,
"reply_to": settings.reply_to,
"logo_url": settings.logo_url,
"primary_color": settings.primary_color,
"secondary_color": settings.secondary_color,
"footer_text": settings.footer_text,
"company_name": settings.company_name,
"company_address": settings.company_address,
}
with translate_domain_errors():
return service.update_settings(tenant_id, body)
@router.get("/logs")
@@ -294,147 +99,58 @@ async def get_send_logs(
offset: int = Query(0, ge=0),
template_type: Optional[str] = Query(None),
tenant_id: str = Depends(_get_tenant),
db: Session = Depends(get_db),
):
service: EmailTemplateService = Depends(get_template_service),
) -> dict[str, Any]:
"""Send-Logs (paginiert)."""
tid = uuid.UUID(tenant_id)
query = db.query(EmailSendLogDB).filter(EmailSendLogDB.tenant_id == tid)
if template_type:
query = query.filter(EmailSendLogDB.template_type == template_type)
total = query.count()
logs = query.order_by(EmailSendLogDB.sent_at.desc()).offset(offset).limit(limit).all()
return {
"logs": [
{
"id": str(l.id),
"template_type": l.template_type,
"recipient": l.recipient,
"subject": l.subject,
"status": l.status,
"variables": l.variables or {},
"error_message": l.error_message,
"sent_at": l.sent_at.isoformat() if l.sent_at else None,
}
for l in logs
],
"total": total,
"limit": limit,
"offset": offset,
}
with translate_domain_errors():
return service.send_logs(tenant_id, limit, offset, template_type)
@router.post("/initialize")
async def initialize_defaults(
tenant_id: str = Depends(_get_tenant),
db: Session = Depends(get_db),
):
service: EmailTemplateService = Depends(get_template_service),
) -> dict[str, Any]:
"""Default-Templates fuer einen Tenant initialisieren."""
tid = uuid.UUID(tenant_id)
existing = db.query(EmailTemplateDB).filter(EmailTemplateDB.tenant_id == tid).count()
if existing > 0:
return {"message": "Templates already initialized", "count": existing}
created = 0
for idx, (ttype, info) in enumerate(TEMPLATE_TYPES.items()):
t = EmailTemplateDB(
tenant_id=tid,
template_type=ttype,
name=info["name"],
category=info["category"],
sort_order=idx * 10,
variables=info["variables"],
)
db.add(t)
created += 1
db.commit()
return {"message": f"{created} templates created", "count": created}
with translate_domain_errors():
return service.initialize_defaults(tenant_id)
@router.get("/default/{template_type}")
async def get_default_content(template_type: str):
async def get_default_content(template_type: str) -> dict[str, Any]:
"""Default-Content fuer einen Template-Typ."""
if template_type not in TEMPLATE_TYPES:
raise HTTPException(status_code=404, detail=f"Unknown template type: {template_type}")
info = TEMPLATE_TYPES[template_type]
vars_html = " ".join([f'<span style="color:#4F46E5">{{{{{v}}}}}</span>' for v in info["variables"]])
return {
"template_type": template_type,
"name": info["name"],
"category": info["category"],
"variables": info["variables"],
"default_subject": f"{info['name']} - {{{{company_name}}}}",
"default_body_html": f"<p>Sehr geehrte(r) {{{{user_name}}}},</p>\n<p>[Inhalt hier einfuegen]</p>\n<p>Verfuegbare Variablen: {vars_html}</p>\n<p>Mit freundlichen Gruessen<br/>{{{{sender_name}}}}</p>",
}
with translate_domain_errors():
return EmailTemplateService.default_content(template_type)
# =============================================================================
# Template CRUD (MUST be before /{id} parameterized routes)
# Template CRUD
# =============================================================================
@router.get("")
async def list_templates(
category: Optional[str] = Query(None),
tenant_id: str = Depends(_get_tenant),
db: Session = Depends(get_db),
):
service: EmailTemplateService = Depends(get_template_service),
) -> list[dict[str, Any]]:
"""Alle Templates mit letzter publizierter Version."""
tid = uuid.UUID(tenant_id)
query = db.query(EmailTemplateDB).filter(EmailTemplateDB.tenant_id == tid)
if category:
query = query.filter(EmailTemplateDB.category == category)
templates = query.order_by(EmailTemplateDB.sort_order).all()
result = []
for t in templates:
latest = db.query(EmailTemplateVersionDB).filter(
EmailTemplateVersionDB.template_id == t.id,
).order_by(EmailTemplateVersionDB.created_at.desc()).first()
result.append(_template_to_dict(t, latest))
return result
with translate_domain_errors():
return service.list_templates(tenant_id, category)
@router.post("")
async def create_template(
body: TemplateCreate,
tenant_id: str = Depends(_get_tenant),
db: Session = Depends(get_db),
):
service: EmailTemplateService = Depends(get_template_service),
) -> dict[str, Any]:
"""Template erstellen."""
if body.template_type not in TEMPLATE_TYPES:
raise HTTPException(status_code=400, detail=f"Unknown template type: {body.template_type}")
tid = uuid.UUID(tenant_id)
existing = db.query(EmailTemplateDB).filter(
EmailTemplateDB.tenant_id == tid,
EmailTemplateDB.template_type == body.template_type,
).first()
if existing:
raise HTTPException(status_code=409, detail=f"Template type '{body.template_type}' already exists")
info = TEMPLATE_TYPES[body.template_type]
t = EmailTemplateDB(
tenant_id=tid,
template_type=body.template_type,
name=body.name or info["name"],
description=body.description,
category=body.category or info["category"],
is_active=body.is_active,
variables=info["variables"],
)
db.add(t)
db.commit()
db.refresh(t)
return _template_to_dict(t)
with translate_domain_errors():
return service.create_template(tenant_id, body)
# =============================================================================
# Version Management (static paths before parameterized)
# Version Management (static path before parameterized)
# =============================================================================
@router.post("/versions")
@@ -442,34 +158,11 @@ async def create_version(
body: VersionCreate,
template_id: str = Query(..., alias="template_id"),
tenant_id: str = Depends(_get_tenant),
db: Session = Depends(get_db),
):
service: EmailTemplateVersionService = Depends(get_version_service),
) -> dict[str, Any]:
"""Neue Version erstellen (via query param template_id)."""
try:
tid = uuid.UUID(template_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid template ID")
template = db.query(EmailTemplateDB).filter(
EmailTemplateDB.id == tid,
EmailTemplateDB.tenant_id == uuid.UUID(tenant_id),
).first()
if not template:
raise HTTPException(status_code=404, detail="Template not found")
v = EmailTemplateVersionDB(
template_id=tid,
version=body.version,
language=body.language,
subject=body.subject,
body_html=body.body_html,
body_text=body.body_text,
status="draft",
)
db.add(v)
db.commit()
db.refresh(v)
return _version_to_dict(v)
with translate_domain_errors():
return service.create_version(tenant_id, template_id, body)
# =============================================================================
@@ -480,52 +173,22 @@ async def create_version(
async def get_template(
template_id: str,
tenant_id: str = Depends(_get_tenant),
db: Session = Depends(get_db),
):
service: EmailTemplateService = Depends(get_template_service),
) -> dict[str, Any]:
"""Template-Detail."""
try:
tid = uuid.UUID(template_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid template ID")
t = db.query(EmailTemplateDB).filter(
EmailTemplateDB.id == tid,
EmailTemplateDB.tenant_id == uuid.UUID(tenant_id),
).first()
if not t:
raise HTTPException(status_code=404, detail="Template not found")
latest = db.query(EmailTemplateVersionDB).filter(
EmailTemplateVersionDB.template_id == t.id,
).order_by(EmailTemplateVersionDB.created_at.desc()).first()
return _template_to_dict(t, latest)
with translate_domain_errors():
return service.get_template(tenant_id, template_id)
@router.get("/{template_id}/versions")
async def get_versions(
template_id: str,
tenant_id: str = Depends(_get_tenant),
db: Session = Depends(get_db),
):
service: EmailTemplateVersionService = Depends(get_version_service),
) -> list[dict[str, Any]]:
"""Versionen eines Templates."""
try:
tid = uuid.UUID(template_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid template ID")
template = db.query(EmailTemplateDB).filter(
EmailTemplateDB.id == tid,
EmailTemplateDB.tenant_id == uuid.UUID(tenant_id),
).first()
if not template:
raise HTTPException(status_code=404, detail="Template not found")
versions = db.query(EmailTemplateVersionDB).filter(
EmailTemplateVersionDB.template_id == tid,
).order_by(EmailTemplateVersionDB.created_at.desc()).all()
return [_version_to_dict(v) for v in versions]
with translate_domain_errors():
return service.list_versions(tenant_id, template_id)
@router.post("/{template_id}/versions")
@@ -533,34 +196,11 @@ async def create_version_for_template(
template_id: str,
body: VersionCreate,
tenant_id: str = Depends(_get_tenant),
db: Session = Depends(get_db),
):
service: EmailTemplateVersionService = Depends(get_version_service),
) -> dict[str, Any]:
"""Neue Version fuer ein Template erstellen."""
try:
tid = uuid.UUID(template_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid template ID")
template = db.query(EmailTemplateDB).filter(
EmailTemplateDB.id == tid,
EmailTemplateDB.tenant_id == uuid.UUID(tenant_id),
).first()
if not template:
raise HTTPException(status_code=404, detail="Template not found")
v = EmailTemplateVersionDB(
template_id=tid,
version=body.version,
language=body.language,
subject=body.subject,
body_html=body.body_html,
body_text=body.body_text,
status="draft",
)
db.add(v)
db.commit()
db.refresh(v)
return _version_to_dict(v)
with translate_domain_errors():
return service.create_version(tenant_id, template_id, body)
# =============================================================================
@@ -570,211 +210,75 @@ async def create_version_for_template(
@router.get("/versions/{version_id}")
async def get_version(
version_id: str,
db: Session = Depends(get_db),
):
service: EmailTemplateVersionService = Depends(get_version_service),
) -> dict[str, Any]:
"""Version-Detail."""
try:
vid = uuid.UUID(version_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid version ID")
v = db.query(EmailTemplateVersionDB).filter(
EmailTemplateVersionDB.id == vid,
).first()
if not v:
raise HTTPException(status_code=404, detail="Version not found")
return _version_to_dict(v)
with translate_domain_errors():
return service.get_version(version_id)
@router.put("/versions/{version_id}")
async def update_version(
version_id: str,
body: VersionUpdate,
db: Session = Depends(get_db),
):
service: EmailTemplateVersionService = Depends(get_version_service),
) -> dict[str, Any]:
"""Draft aktualisieren."""
try:
vid = uuid.UUID(version_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid version ID")
v = db.query(EmailTemplateVersionDB).filter(
EmailTemplateVersionDB.id == vid,
).first()
if not v:
raise HTTPException(status_code=404, detail="Version not found")
if v.status != "draft":
raise HTTPException(status_code=400, detail="Only draft versions can be edited")
if body.subject is not None:
v.subject = body.subject
if body.body_html is not None:
v.body_html = body.body_html
if body.body_text is not None:
v.body_text = body.body_text
db.commit()
db.refresh(v)
return _version_to_dict(v)
with translate_domain_errors():
return service.update_version(version_id, body)
@router.post("/versions/{version_id}/submit")
async def submit_version(
version_id: str,
db: Session = Depends(get_db),
):
service: EmailTemplateVersionService = Depends(get_version_service),
) -> dict[str, Any]:
"""Zur Pruefung einreichen."""
try:
vid = uuid.UUID(version_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid version ID")
v = db.query(EmailTemplateVersionDB).filter(
EmailTemplateVersionDB.id == vid,
).first()
if not v:
raise HTTPException(status_code=404, detail="Version not found")
if v.status != "draft":
raise HTTPException(status_code=400, detail="Only draft versions can be submitted")
v.status = "review"
v.submitted_at = datetime.now(timezone.utc)
v.submitted_by = "admin"
db.commit()
db.refresh(v)
return _version_to_dict(v)
with translate_domain_errors():
return service.submit(version_id)
@router.post("/versions/{version_id}/approve")
async def approve_version(
version_id: str,
comment: Optional[str] = None,
db: Session = Depends(get_db),
):
service: EmailTemplateVersionService = Depends(get_version_service),
) -> dict[str, Any]:
"""Genehmigen."""
try:
vid = uuid.UUID(version_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid version ID")
v = db.query(EmailTemplateVersionDB).filter(
EmailTemplateVersionDB.id == vid,
).first()
if not v:
raise HTTPException(status_code=404, detail="Version not found")
if v.status != "review":
raise HTTPException(status_code=400, detail="Only review versions can be approved")
v.status = "approved"
approval = EmailTemplateApprovalDB(
version_id=vid,
action="approve",
comment=comment,
approved_by="admin",
)
db.add(approval)
db.commit()
db.refresh(v)
return _version_to_dict(v)
with translate_domain_errors():
return service.approve(version_id, comment)
@router.post("/versions/{version_id}/reject")
async def reject_version(
version_id: str,
comment: Optional[str] = None,
db: Session = Depends(get_db),
):
service: EmailTemplateVersionService = Depends(get_version_service),
) -> dict[str, Any]:
"""Ablehnen."""
try:
vid = uuid.UUID(version_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid version ID")
v = db.query(EmailTemplateVersionDB).filter(
EmailTemplateVersionDB.id == vid,
).first()
if not v:
raise HTTPException(status_code=404, detail="Version not found")
if v.status != "review":
raise HTTPException(status_code=400, detail="Only review versions can be rejected")
v.status = "draft" # Back to draft
approval = EmailTemplateApprovalDB(
version_id=vid,
action="reject",
comment=comment,
approved_by="admin",
)
db.add(approval)
db.commit()
db.refresh(v)
return _version_to_dict(v)
with translate_domain_errors():
return service.reject(version_id, comment)
@router.post("/versions/{version_id}/publish")
async def publish_version(
version_id: str,
db: Session = Depends(get_db),
):
service: EmailTemplateVersionService = Depends(get_version_service),
) -> dict[str, Any]:
"""Publizieren."""
try:
vid = uuid.UUID(version_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid version ID")
v = db.query(EmailTemplateVersionDB).filter(
EmailTemplateVersionDB.id == vid,
).first()
if not v:
raise HTTPException(status_code=404, detail="Version not found")
if v.status not in ("approved", "review", "draft"):
raise HTTPException(status_code=400, detail="Version cannot be published")
now = datetime.now(timezone.utc)
v.status = "published"
v.published_at = now
v.published_by = "admin"
db.commit()
db.refresh(v)
return _version_to_dict(v)
with translate_domain_errors():
return service.publish(version_id)
@router.post("/versions/{version_id}/preview")
async def preview_version(
version_id: str,
body: PreviewRequest,
db: Session = Depends(get_db),
):
service: EmailTemplateVersionService = Depends(get_version_service),
) -> dict[str, Any]:
"""Vorschau mit Test-Variablen."""
try:
vid = uuid.UUID(version_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid version ID")
v = db.query(EmailTemplateVersionDB).filter(
EmailTemplateVersionDB.id == vid,
).first()
if not v:
raise HTTPException(status_code=404, detail="Version not found")
variables = body.variables or {}
# Fill in defaults for missing variables
template = db.query(EmailTemplateDB).filter(
EmailTemplateDB.id == v.template_id,
).first()
if template and template.variables:
for var in template.variables:
if var not in variables:
variables[var] = f"[{var}]"
rendered_subject = _render_template(v.subject, variables)
rendered_html = _render_template(v.body_html, variables)
return {
"subject": rendered_subject,
"body_html": rendered_html,
"variables_used": variables,
}
with translate_domain_errors():
return service.preview(version_id, body)
@router.post("/versions/{version_id}/send-test")
@@ -782,42 +286,17 @@ async def send_test_email(
version_id: str,
body: SendTestRequest,
tenant_id: str = Depends(_get_tenant),
db: Session = Depends(get_db),
):
service: EmailTemplateVersionService = Depends(get_version_service),
) -> dict[str, Any]:
"""Test-E-Mail senden (Simulation — loggt nur)."""
try:
vid = uuid.UUID(version_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid version ID")
with translate_domain_errors():
return service.send_test(tenant_id, version_id, body)
v = db.query(EmailTemplateVersionDB).filter(
EmailTemplateVersionDB.id == vid,
).first()
if not v:
raise HTTPException(status_code=404, detail="Version not found")
template = db.query(EmailTemplateDB).filter(
EmailTemplateDB.id == v.template_id,
).first()
variables = body.variables or {}
rendered_subject = _render_template(v.subject, variables)
# Log the send attempt
log = EmailSendLogDB(
tenant_id=uuid.UUID(tenant_id),
template_type=template.template_type if template else "unknown",
version_id=vid,
recipient=body.recipient,
subject=rendered_subject,
status="test_sent",
variables=variables,
)
db.add(log)
db.commit()
return {
"success": True,
"message": f"Test-E-Mail an {body.recipient} gesendet (Simulation)",
"subject": rendered_subject,
}
# Legacy re-exports
__all__ = [
"router",
"TEMPLATE_TYPES",
"VALID_CATEGORIES",
"VALID_STATUSES",
]

62
backend-compliance/compliance/schemas/email_template.py Normal file
View File

@@ -0,0 +1,62 @@
"""
Email template request schemas.
Phase 1 Step 4: extracted from ``compliance.api.email_template_routes``.
"""
from typing import Optional
from pydantic import BaseModel
class TemplateCreate(BaseModel):
template_type: str
name: Optional[str] = None
description: Optional[str] = None
category: Optional[str] = None
is_active: bool = True
class VersionCreate(BaseModel):
version: str = "1.0"
language: str = "de"
subject: str
body_html: str
body_text: Optional[str] = None
class VersionUpdate(BaseModel):
subject: Optional[str] = None
body_html: Optional[str] = None
body_text: Optional[str] = None
class PreviewRequest(BaseModel):
variables: Optional[dict[str, str]] = None
class SendTestRequest(BaseModel):
recipient: str
variables: Optional[dict[str, str]] = None
class SettingsUpdate(BaseModel):
sender_name: Optional[str] = None
sender_email: Optional[str] = None
reply_to: Optional[str] = None
logo_url: Optional[str] = None
primary_color: Optional[str] = None
secondary_color: Optional[str] = None
footer_text: Optional[str] = None
company_name: Optional[str] = None
company_address: Optional[str] = None
__all__ = [
"TemplateCreate",
"VersionCreate",
"VersionUpdate",
"PreviewRequest",
"SendTestRequest",
"SettingsUpdate",
]

397
backend-compliance/compliance/services/email_template_service.py Normal file
View File

@@ -0,0 +1,397 @@
# mypy: disable-error-code="arg-type,assignment,union-attr"
"""
Email Template service — templates CRUD, settings, logs, stats, initialization.
Phase 1 Step 4: extracted from ``compliance.api.email_template_routes``.
Version workflow (draft/review/approve/reject/publish/preview/send-test)
lives in ``compliance.services.email_template_version_service``.
"""
import uuid
from datetime import datetime, timezone
from typing import Any, Optional
from sqlalchemy.orm import Session
from compliance.db.email_template_models import (
EmailSendLogDB,
EmailTemplateDB,
EmailTemplateSettingsDB,
EmailTemplateVersionDB,
)
from compliance.domain import ConflictError, NotFoundError, ValidationError
from compliance.schemas.email_template import SettingsUpdate, TemplateCreate
# Template type catalog — shared across both services and the route module.
TEMPLATE_TYPES: dict[str, dict[str, Any]] = {
"welcome": {"name": "Willkommen", "category": "general", "variables": ["user_name", "company_name", "login_url"]},
"verification": {"name": "E-Mail-Verifizierung", "category": "general", "variables": ["user_name", "verification_url", "expiry_hours"]},
"password_reset": {"name": "Passwort zuruecksetzen", "category": "general", "variables": ["user_name", "reset_url", "expiry_hours"]},
"dsr_receipt": {"name": "DSR Eingangsbestaetigung", "category": "dsr", "variables": ["requester_name", "reference_number", "request_type", "deadline"]},
"dsr_identity_request": {"name": "DSR Identitaetsanfrage", "category": "dsr", "variables": ["requester_name", "reference_number"]},
"dsr_completion": {"name": "DSR Abschluss", "category": "dsr", "variables": ["requester_name", "reference_number", "request_type", "completion_date"]},
"dsr_rejection": {"name": "DSR Ablehnung", "category": "dsr", "variables": ["requester_name", "reference_number", "rejection_reason", "legal_basis"]},
"dsr_extension": {"name": "DSR Fristverlaengerung", "category": "dsr", "variables": ["requester_name", "reference_number", "new_deadline", "extension_reason"]},
"consent_request": {"name": "Einwilligungsanfrage", "category": "consent", "variables": ["user_name", "purpose", "consent_url"]},
"consent_confirmation": {"name": "Einwilligungsbestaetigung", "category": "consent", "variables": ["user_name", "purpose", "consent_date"]},
"consent_withdrawal": {"name": "Widerruf bestaetigt", "category": "consent", "variables": ["user_name", "purpose", "withdrawal_date"]},
"consent_reminder": {"name": "Einwilligungs-Erinnerung", "category": "consent", "variables": ["user_name", "purpose", "expiry_date"]},
"breach_notification_authority": {"name": "Datenpanne Aufsichtsbehoerde", "category": "breach", "variables": ["incident_date", "incident_description", "affected_count", "measures_taken", "authority_name"]},
"breach_notification_affected": {"name": "Datenpanne Betroffene", "category": "breach", "variables": ["user_name", "incident_date", "incident_description", "measures_taken", "contact_info"]},
"breach_internal": {"name": "Datenpanne intern", "category": "breach", "variables": ["reporter_name", "incident_date", "incident_description", "severity"]},
"vendor_dpa_request": {"name": "AVV-Anfrage", "category": "vendor", "variables": ["vendor_name", "contact_name", "deadline", "requirements"]},
"vendor_review_reminder": {"name": "Vendor-Pruefung Erinnerung", "category": "vendor", "variables": ["vendor_name", "review_due_date", "last_review_date"]},
"training_invitation": {"name": "Schulungseinladung", "category": "training", "variables": ["user_name", "training_title", "training_date", "training_url"]},
"training_reminder": {"name": "Schulungs-Erinnerung", "category": "training", "variables": ["user_name", "training_title", "deadline"]},
"training_completion": {"name": "Schulung abgeschlossen", "category": "training", "variables": ["user_name", "training_title", "completion_date", "certificate_url"]},
}
VALID_STATUSES = ["draft", "review", "approved", "published"]
VALID_CATEGORIES = ["general", "dsr", "consent", "breach", "vendor", "training"]
def _template_to_dict(
t: EmailTemplateDB, latest_version: Optional[EmailTemplateVersionDB] = None
) -> dict[str, Any]:
result: dict[str, Any] = {
"id": str(t.id),
"tenant_id": str(t.tenant_id),
"template_type": t.template_type,
"name": t.name,
"description": t.description,
"category": t.category,
"is_active": t.is_active,
"sort_order": t.sort_order,
"variables": t.variables or [],
"created_at": t.created_at.isoformat() if t.created_at else None,
"updated_at": t.updated_at.isoformat() if t.updated_at else None,
}
if latest_version:
result["latest_version"] = _version_to_dict(latest_version)
return result
def _version_to_dict(v: EmailTemplateVersionDB) -> dict[str, Any]:
return {
"id": str(v.id),
"template_id": str(v.template_id),
"version": v.version,
"language": v.language,
"subject": v.subject,
"body_html": v.body_html,
"body_text": v.body_text,
"status": v.status,
"submitted_at": v.submitted_at.isoformat() if v.submitted_at else None,
"submitted_by": v.submitted_by,
"published_at": v.published_at.isoformat() if v.published_at else None,
"published_by": v.published_by,
"created_at": v.created_at.isoformat() if v.created_at else None,
"created_by": v.created_by,
}
def _render_template(html: str, variables: dict[str, str]) -> str:
"""Replace {{variable}} placeholders with values."""
result = html
for key, value in variables.items():
result = result.replace(f"{{{{{key}}}}}", str(value))
return result
class EmailTemplateService:
"""Business logic for templates, settings, logs, stats, initialization."""
def __init__(self, db: Session) -> None:
self.db = db
# ------------------------------------------------------------------
# Type catalog + defaults
# ------------------------------------------------------------------
@staticmethod
def list_types() -> list[dict[str, Any]]:
return [
{
"type": ttype,
"name": info["name"],
"category": info["category"],
"variables": info["variables"],
}
for ttype, info in TEMPLATE_TYPES.items()
]
@staticmethod
def default_content(template_type: str) -> dict[str, Any]:
if template_type not in TEMPLATE_TYPES:
raise NotFoundError(f"Unknown template type: {template_type}")
info = TEMPLATE_TYPES[template_type]
vars_html = " ".join(
f'<span style="color:#4F46E5">{{{{{v}}}}}</span>' for v in info["variables"]
)
return {
"template_type": template_type,
"name": info["name"],
"category": info["category"],
"variables": info["variables"],
"default_subject": f"{info['name']} - {{{{company_name}}}}",
"default_body_html": (
f"<p>Sehr geehrte(r) {{{{user_name}}}},</p>\n"
f"<p>[Inhalt hier einfuegen]</p>\n"
f"<p>Verfuegbare Variablen: {vars_html}</p>\n"
f"<p>Mit freundlichen Gruessen<br/>{{{{sender_name}}}}</p>"
),
}
# ------------------------------------------------------------------
# Stats
# ------------------------------------------------------------------
def stats(self, tenant_id: str) -> dict[str, Any]:
tid = uuid.UUID(tenant_id)
base = self.db.query(EmailTemplateDB).filter(EmailTemplateDB.tenant_id == tid)
total = base.count()
active = base.filter(EmailTemplateDB.is_active).count()
published_count = 0
for t in base.all():
has_published = (
self.db.query(EmailTemplateVersionDB)
.filter(
EmailTemplateVersionDB.template_id == t.id,
EmailTemplateVersionDB.status == "published",
)
.count()
> 0
)
if has_published:
published_count += 1
by_category: dict[str, int] = {}
for cat in VALID_CATEGORIES:
by_category[cat] = base.filter(EmailTemplateDB.category == cat).count()
total_sent = (
self.db.query(EmailSendLogDB)
.filter(EmailSendLogDB.tenant_id == tid)
.count()
)
return {
"total": total,
"active": active,
"published": published_count,
"draft": total - published_count,
"by_category": by_category,
"total_sent": total_sent,
}
# ------------------------------------------------------------------
# Settings
# ------------------------------------------------------------------
@staticmethod
def _settings_defaults() -> dict[str, Any]:
return {
"sender_name": "Datenschutzbeauftragter",
"sender_email": "datenschutz@example.de",
"reply_to": None,
"logo_url": None,
"primary_color": "#4F46E5",
"secondary_color": "#7C3AED",
"footer_text": "Datenschutzhinweis: Diese E-Mail enthaelt vertrauliche Informationen.",
"company_name": None,
"company_address": None,
}
@staticmethod
def _settings_to_dict(s: EmailTemplateSettingsDB) -> dict[str, Any]:
return {
"sender_name": s.sender_name,
"sender_email": s.sender_email,
"reply_to": s.reply_to,
"logo_url": s.logo_url,
"primary_color": s.primary_color,
"secondary_color": s.secondary_color,
"footer_text": s.footer_text,
"company_name": s.company_name,
"company_address": s.company_address,
}
def get_settings(self, tenant_id: str) -> dict[str, Any]:
tid = uuid.UUID(tenant_id)
settings = (
self.db.query(EmailTemplateSettingsDB)
.filter(EmailTemplateSettingsDB.tenant_id == tid)
.first()
)
if not settings:
return self._settings_defaults()
return self._settings_to_dict(settings)
def update_settings(self, tenant_id: str, body: SettingsUpdate) -> dict[str, Any]:
tid = uuid.UUID(tenant_id)
settings = (
self.db.query(EmailTemplateSettingsDB)
.filter(EmailTemplateSettingsDB.tenant_id == tid)
.first()
)
if not settings:
settings = EmailTemplateSettingsDB(tenant_id=tid)
self.db.add(settings)
for field in (
"sender_name", "sender_email", "reply_to", "logo_url",
"primary_color", "secondary_color", "footer_text",
"company_name", "company_address",
):
val = getattr(body, field, None)
if val is not None:
setattr(settings, field, val)
settings.updated_at = datetime.now(timezone.utc)
self.db.commit()
self.db.refresh(settings)
return self._settings_to_dict(settings)
# ------------------------------------------------------------------
# Send logs
# ------------------------------------------------------------------
def send_logs(
self,
tenant_id: str,
limit: int,
offset: int,
template_type: Optional[str],
) -> dict[str, Any]:
tid = uuid.UUID(tenant_id)
q = self.db.query(EmailSendLogDB).filter(EmailSendLogDB.tenant_id == tid)
if template_type:
q = q.filter(EmailSendLogDB.template_type == template_type)
total = q.count()
logs = (
q.order_by(EmailSendLogDB.sent_at.desc())
.offset(offset)
.limit(limit)
.all()
)
return {
"logs": [
{
"id": str(lg.id),
"template_type": lg.template_type,
"recipient": lg.recipient,
"subject": lg.subject,
"status": lg.status,
"variables": lg.variables or {},
"error_message": lg.error_message,
"sent_at": lg.sent_at.isoformat() if lg.sent_at else None,
}
for lg in logs
],
"total": total,
"limit": limit,
"offset": offset,
}
# ------------------------------------------------------------------
# Initialization + template CRUD
# ------------------------------------------------------------------
def initialize_defaults(self, tenant_id: str) -> dict[str, Any]:
tid = uuid.UUID(tenant_id)
existing = (
self.db.query(EmailTemplateDB)
.filter(EmailTemplateDB.tenant_id == tid)
.count()
)
if existing > 0:
return {"message": "Templates already initialized", "count": existing}
created = 0
for idx, (ttype, info) in enumerate(TEMPLATE_TYPES.items()):
self.db.add(
EmailTemplateDB(
tenant_id=tid,
template_type=ttype,
name=info["name"],
category=info["category"],
sort_order=idx * 10,
variables=info["variables"],
)
)
created += 1
self.db.commit()
return {"message": f"{created} templates created", "count": created}
def list_templates(
self, tenant_id: str, category: Optional[str]
) -> list[dict[str, Any]]:
tid = uuid.UUID(tenant_id)
q = self.db.query(EmailTemplateDB).filter(EmailTemplateDB.tenant_id == tid)
if category:
q = q.filter(EmailTemplateDB.category == category)
templates = q.order_by(EmailTemplateDB.sort_order).all()
result = []
for t in templates:
latest = (
self.db.query(EmailTemplateVersionDB)
.filter(EmailTemplateVersionDB.template_id == t.id)
.order_by(EmailTemplateVersionDB.created_at.desc())
.first()
)
result.append(_template_to_dict(t, latest))
return result
def create_template(self, tenant_id: str, body: TemplateCreate) -> dict[str, Any]:
if body.template_type not in TEMPLATE_TYPES:
raise ValidationError(f"Unknown template type: {body.template_type}")
tid = uuid.UUID(tenant_id)
existing = (
self.db.query(EmailTemplateDB)
.filter(
EmailTemplateDB.tenant_id == tid,
EmailTemplateDB.template_type == body.template_type,
)
.first()
)
if existing:
raise ConflictError(
f"Template type '{body.template_type}' already exists"
)
info = TEMPLATE_TYPES[body.template_type]
t = EmailTemplateDB(
tenant_id=tid,
template_type=body.template_type,
name=body.name or info["name"],
description=body.description,
category=body.category or info["category"],
is_active=body.is_active,
variables=info["variables"],
)
self.db.add(t)
self.db.commit()
self.db.refresh(t)
return _template_to_dict(t)
def get_template(self, tenant_id: str, template_id: str) -> dict[str, Any]:
try:
tid = uuid.UUID(template_id)
except ValueError as exc:
raise ValidationError("Invalid template ID") from exc
t = (
self.db.query(EmailTemplateDB)
.filter(
EmailTemplateDB.id == tid,
EmailTemplateDB.tenant_id == uuid.UUID(tenant_id),
)
.first()
)
if not t:
raise NotFoundError("Template not found")
latest = (
self.db.query(EmailTemplateVersionDB)
.filter(EmailTemplateVersionDB.template_id == t.id)
.order_by(EmailTemplateVersionDB.created_at.desc())
.first()
)
return _template_to_dict(t, latest)

260
backend-compliance/compliance/services/email_template_version_service.py Normal file
View File

@@ -0,0 +1,260 @@
# mypy: disable-error-code="arg-type,assignment,union-attr"
"""
Email Template version service — version workflow (draft/review/approve/
reject/publish/preview/send-test).
Phase 1 Step 4: extracted from ``compliance.api.email_template_routes``.
Template-level CRUD + settings + stats live in
``compliance.services.email_template_service``.
"""
import uuid
from datetime import datetime, timezone
from typing import Any, Optional
from sqlalchemy.orm import Session
from compliance.db.email_template_models import (
EmailSendLogDB,
EmailTemplateApprovalDB,
EmailTemplateDB,
EmailTemplateVersionDB,
)
from compliance.domain import ConflictError, NotFoundError, ValidationError
from compliance.schemas.email_template import (
PreviewRequest,
SendTestRequest,
VersionCreate,
VersionUpdate,
)
from compliance.services.email_template_service import (
_render_template,
_version_to_dict,
)
class EmailTemplateVersionService:
"""Business logic for email-template version workflow + preview + test-send."""
def __init__(self, db: Session) -> None:
self.db = db
# ------------------------------------------------------------------
# Internal lookups
# ------------------------------------------------------------------
@staticmethod
def _parse_template_uuid(template_id: str) -> uuid.UUID:
try:
return uuid.UUID(template_id)
except ValueError as exc:
raise ValidationError("Invalid template ID") from exc
@staticmethod
def _parse_version_uuid(version_id: str) -> uuid.UUID:
try:
return uuid.UUID(version_id)
except ValueError as exc:
raise ValidationError("Invalid version ID") from exc
def _template_or_raise(
self, tenant_id: str, tid: uuid.UUID
) -> EmailTemplateDB:
template = (
self.db.query(EmailTemplateDB)
.filter(
EmailTemplateDB.id == tid,
EmailTemplateDB.tenant_id == uuid.UUID(tenant_id),
)
.first()
)
if not template:
raise NotFoundError("Template not found")
return template
def _version_or_raise(self, vid: uuid.UUID) -> EmailTemplateVersionDB:
v = (
self.db.query(EmailTemplateVersionDB)
.filter(EmailTemplateVersionDB.id == vid)
.first()
)
if not v:
raise NotFoundError("Version not found")
return v
# ------------------------------------------------------------------
# Create / list / get versions
# ------------------------------------------------------------------
def create_version(
self, tenant_id: str, template_id: str, body: VersionCreate
) -> dict[str, Any]:
tid = self._parse_template_uuid(template_id)
self._template_or_raise(tenant_id, tid)
v = EmailTemplateVersionDB(
template_id=tid,
version=body.version,
language=body.language,
subject=body.subject,
body_html=body.body_html,
body_text=body.body_text,
status="draft",
)
self.db.add(v)
self.db.commit()
self.db.refresh(v)
return _version_to_dict(v)
def list_versions(
self, tenant_id: str, template_id: str
) -> list[dict[str, Any]]:
tid = self._parse_template_uuid(template_id)
self._template_or_raise(tenant_id, tid)
versions = (
self.db.query(EmailTemplateVersionDB)
.filter(EmailTemplateVersionDB.template_id == tid)
.order_by(EmailTemplateVersionDB.created_at.desc())
.all()
)
return [_version_to_dict(v) for v in versions]
def get_version(self, version_id: str) -> dict[str, Any]:
vid = self._parse_version_uuid(version_id)
return _version_to_dict(self._version_or_raise(vid))
def update_version(
self, version_id: str, body: VersionUpdate
) -> dict[str, Any]:
vid = self._parse_version_uuid(version_id)
v = self._version_or_raise(vid)
if v.status != "draft":
raise ValidationError("Only draft versions can be edited")
if body.subject is not None:
v.subject = body.subject
if body.body_html is not None:
v.body_html = body.body_html
if body.body_text is not None:
v.body_text = body.body_text
self.db.commit()
self.db.refresh(v)
return _version_to_dict(v)
# ------------------------------------------------------------------
# Workflow transitions
# ------------------------------------------------------------------
def submit(self, version_id: str) -> dict[str, Any]:
vid = self._parse_version_uuid(version_id)
v = self._version_or_raise(vid)
if v.status != "draft":
raise ValidationError("Only draft versions can be submitted")
v.status = "review"
v.submitted_at = datetime.now(timezone.utc)
v.submitted_by = "admin"
self.db.commit()
self.db.refresh(v)
return _version_to_dict(v)
def approve(self, version_id: str, comment: Optional[str]) -> dict[str, Any]:
vid = self._parse_version_uuid(version_id)
v = self._version_or_raise(vid)
if v.status != "review":
raise ValidationError("Only review versions can be approved")
v.status = "approved"
self.db.add(
EmailTemplateApprovalDB(
version_id=vid,
action="approve",
comment=comment,
approved_by="admin",
)
)
self.db.commit()
self.db.refresh(v)
return _version_to_dict(v)
def reject(self, version_id: str, comment: Optional[str]) -> dict[str, Any]:
vid = self._parse_version_uuid(version_id)
v = self._version_or_raise(vid)
if v.status != "review":
raise ValidationError("Only review versions can be rejected")
v.status = "draft"
self.db.add(
EmailTemplateApprovalDB(
version_id=vid,
action="reject",
comment=comment,
approved_by="admin",
)
)
self.db.commit()
self.db.refresh(v)
return _version_to_dict(v)
def publish(self, version_id: str) -> dict[str, Any]:
vid = self._parse_version_uuid(version_id)
v = self._version_or_raise(vid)
if v.status not in ("approved", "review", "draft"):
raise ValidationError("Version cannot be published")
v.status = "published"
v.published_at = datetime.now(timezone.utc)
v.published_by = "admin"
self.db.commit()
self.db.refresh(v)
return _version_to_dict(v)
# ------------------------------------------------------------------
# Preview + test send
# ------------------------------------------------------------------
def preview(self, version_id: str, body: PreviewRequest) -> dict[str, Any]:
vid = self._parse_version_uuid(version_id)
v = self._version_or_raise(vid)
variables = dict(body.variables or {})
template = (
self.db.query(EmailTemplateDB)
.filter(EmailTemplateDB.id == v.template_id)
.first()
)
if template and template.variables:
for var in list(template.variables):
if var not in variables:
variables[var] = f"[{var}]"
return {
"subject": _render_template(v.subject, variables),
"body_html": _render_template(v.body_html, variables),
"variables_used": variables,
}
def send_test(
self, tenant_id: str, version_id: str, body: SendTestRequest
) -> dict[str, Any]:
vid = self._parse_version_uuid(version_id)
v = self._version_or_raise(vid)
template = (
self.db.query(EmailTemplateDB)
.filter(EmailTemplateDB.id == v.template_id)
.first()
)
variables = body.variables or {}
rendered_subject = _render_template(v.subject, variables)
self.db.add(
EmailSendLogDB(
tenant_id=uuid.UUID(tenant_id),
template_type=template.template_type if template else "unknown",
version_id=vid,
recipient=body.recipient,
subject=rendered_subject,
status="test_sent",
variables=variables,
)
)
self.db.commit()
return {
"success": True,
"message": f"Test-E-Mail an {body.recipient} gesendet (Simulation)",
"subject": rendered_subject,
}

2
backend-compliance/mypy.ini
View File

@@ -89,5 +89,7 @@ ignore_errors = False
ignore_errors = False
[mypy-compliance.api.evidence_routes]
ignore_errors = False
[mypy-compliance.api.email_template_routes]
ignore_errors = False
[mypy-compliance.api._http_errors]
ignore_errors = False

401
backend-compliance/tests/contracts/openapi.baseline.json
View File

@@ -19563,135 +19563,6 @@
"title": "ConsentCreate",
"type": "object"
},
"compliance__api__email_template_routes__TemplateCreate": {
"properties": {
"category": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "Category"
},
"description": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "Description"
},
"is_active": {
"default": true,
"title": "Is Active",
"type": "boolean"
},
"name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "Name"
},
"template_type": {
"title": "Template Type",
"type": "string"
}
},
"required": [
"template_type"
],
"title": "TemplateCreate",
"type": "object"
},
"compliance__api__email_template_routes__VersionCreate": {
"properties": {
"body_html": {
"title": "Body Html",
"type": "string"
},
"body_text": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "Body Text"
},
"language": {
"default": "de",
"title": "Language",
"type": "string"
},
"subject": {
"title": "Subject",
"type": "string"
},
"version": {
"default": "1.0",
"title": "Version",
"type": "string"
}
},
"required": [
"subject",
"body_html"
],
"title": "VersionCreate",
"type": "object"
},
"compliance__api__email_template_routes__VersionUpdate": {
"properties": {
"body_html": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "Body Html"
},
"body_text": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "Body Text"
},
"subject": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "Subject"
}
},
"title": "VersionUpdate",
"type": "object"
},
"compliance__api__incident_routes__IncidentCreate": {
"properties": {
"affected_data_categories": {
@@ -20361,6 +20232,135 @@
],
"title": "ConsentCreate",
"type": "object"
},
"compliance__schemas__email_template__TemplateCreate": {
"properties": {
"category": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "Category"
},
"description": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "Description"
},
"is_active": {
"default": true,
"title": "Is Active",
"type": "boolean"
},
"name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "Name"
},
"template_type": {
"title": "Template Type",
"type": "string"
}
},
"required": [
"template_type"
],
"title": "TemplateCreate",
"type": "object"
},
"compliance__schemas__email_template__VersionCreate": {
"properties": {
"body_html": {
"title": "Body Html",
"type": "string"
},
"body_text": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "Body Text"
},
"language": {
"default": "de",
"title": "Language",
"type": "string"
},
"subject": {
"title": "Subject",
"type": "string"
},
"version": {
"default": "1.0",
"title": "Version",
"type": "string"
}
},
"required": [
"subject",
"body_html"
],
"title": "VersionCreate",
"type": "object"
},
"compliance__schemas__email_template__VersionUpdate": {
"properties": {
"body_html": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "Body Html"
},
"body_text": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "Body Text"
},
"subject": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "Subject"
}
},
"title": "VersionUpdate",
"type": "object"
}
}
},
@@ -27260,7 +27260,14 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"items": {
"additionalProperties": true,
"type": "object"
},
"title": "Response List Templates Api Compliance Email Templates Get",
"type": "array"
}
}
},
"description": "Successful Response"
@@ -27307,7 +27314,7 @@
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/compliance__api__email_template_routes__TemplateCreate"
"$ref": "#/components/schemas/compliance__schemas__email_template__TemplateCreate"
}
}
},
@@ -27317,7 +27324,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Create Template Api Compliance Email Templates Post",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -27359,7 +27370,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Get Default Content Api Compliance Email Templates Default Template Type Get",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -27408,7 +27423,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Initialize Defaults Api Compliance Email Templates Initialize Post",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -27496,7 +27515,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Get Send Logs Api Compliance Email Templates Logs Get",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -27545,7 +27568,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Get Settings Api Compliance Email Templates Settings Get",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -27602,7 +27629,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Update Settings Api Compliance Email Templates Settings Put",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -27651,7 +27682,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Get Stats Api Compliance Email Templates Stats Get",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -27682,7 +27717,14 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"items": {
"additionalProperties": true,
"type": "object"
},
"title": "Response Get Template Types Api Compliance Email Templates Types Get",
"type": "array"
}
}
},
"description": "Successful Response"
@@ -27730,7 +27772,7 @@
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/compliance__api__email_template_routes__VersionCreate"
"$ref": "#/components/schemas/compliance__schemas__email_template__VersionCreate"
}
}
},
@@ -27740,7 +27782,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Create Version Api Compliance Email Templates Versions Post",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -27782,7 +27828,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Get Version Api Compliance Email Templates Versions Version Id Get",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -27822,7 +27872,7 @@
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/compliance__api__email_template_routes__VersionUpdate"
"$ref": "#/components/schemas/compliance__schemas__email_template__VersionUpdate"
}
}
},
@@ -27832,7 +27882,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Update Version Api Compliance Email Templates Versions Version Id Put",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -27890,7 +27944,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Approve Version Api Compliance Email Templates Versions Version Id Approve Post",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -27942,7 +28000,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Preview Version Api Compliance Email Templates Versions Version Id Preview Post",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -27984,7 +28046,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Publish Version Api Compliance Email Templates Versions Version Id Publish Post",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -28042,7 +28108,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Reject Version Api Compliance Email Templates Versions Version Id Reject Post",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -28110,7 +28180,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Send Test Email Api Compliance Email Templates Versions Version Id Send Test Post",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -28152,7 +28226,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Submit Version Api Compliance Email Templates Versions Version Id Submit Post",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -28210,7 +28288,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Get Template Api Compliance Email Templates Template Id Get",
"type": "object"
}
}
},
"description": "Successful Response"
@@ -28268,7 +28350,14 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"items": {
"additionalProperties": true,
"type": "object"
},
"title": "Response Get Versions Api Compliance Email Templates Template Id Versions Get",
"type": "array"
}
}
},
"description": "Successful Response"
@@ -28324,7 +28413,7 @@
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/compliance__api__email_template_routes__VersionCreate"
"$ref": "#/components/schemas/compliance__schemas__email_template__VersionCreate"
}
}
},
@@ -28334,7 +28423,11 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"additionalProperties": true,
"title": "Response Create Version For Template Api Compliance Email Templates Template Id Versions Post",
"type": "object"
}
}
},
"description": "Successful Response"