from dataclasses import dataclass from email_campaigns.adapters.db import Repository from email_campaigns.config import settings from email_campaigns.emails.cache import EmailPublicPreviewCache from email_campaigns.emails.exceptions import ( AutosaveUnauthorizedError, EmailNotFoundError, ) from email_campaigns.emails.models import BaseEmail @dataclass(kw_only=True) class AutosaveEmailRequest: key: str email_id: str html: str css: str username: str password: str user_id: str | None = None class AutosaveEmailHandler[T: BaseEmail]: def __init__( self, email_repository: Repository[T], preview_cache: EmailPublicPreviewCache, ) -> None: self.email_repository = email_repository self.preview_cache = preview_cache def handle(self, request: AutosaveEmailRequest) -> None: if ( settings.stripo_autosave_username != request.username or settings.stripo_autosave_password != request.password ): raise AutosaveUnauthorizedError email = self.email_repository.get(request.email_id) if email is None: raise EmailNotFoundError # Extra email validation self.validate_email(email) email.html_content = request.html email.css_content = request.css if request.user_id: email.updated_by = request.user_id self.email_repository.save(email) # Invalidate cache after autosave self.preview_cache.delete(email.id, email.email_type) def validate_email(self, email: T) -> None: """Extra email validation"""