from dataclasses import dataclass from anydi import singleton from fansifter_common.auth.requests import AuthRequest from fansifter_common.auth.services import AuthService from fansifter_common.auth.types import Permission from email_campaigns.adapters.db.decorators import transactional from email_campaigns.automated.models import AutomatedEmail from email_campaigns.automated.repositories import AutomatedEmailRepository from email_campaigns.automated.validators import AutomatedEmailValidator from email_campaigns.emails.exceptions import EmailNotFoundError from email_campaigns.emails.validators import EmailValidator @dataclass(kw_only=True) class DuplicateAutomatedEmailRequest(AuthRequest): email_id: str name: str @singleton class DuplicateAutomatedEmailHandler: permission = Permission("email_campaign", "edit") def __init__( self, auth_service: AuthService, automated_email_repository: AutomatedEmailRepository, automated_email_validator: AutomatedEmailValidator, email_validator: EmailValidator, ) -> None: self.auth_service = auth_service self.automated_email_repository = automated_email_repository self.automated_email_validator = automated_email_validator self.email_validator = email_validator @transactional def handle(self, request: DuplicateAutomatedEmailRequest) -> AutomatedEmail: automated_email = self.automated_email_repository.get(request.email_id) if automated_email is None: raise EmailNotFoundError self.auth_service.authorize_account( request.identity_id, account=automated_email.account, permission=self.permission, ) # Validate the new name self.automated_email_validator.validate_name( request.name, account=automated_email.account ) # Validate required content blocks of the source email before copying if automated_email.html_content is not None: self.email_validator.validate_privacy_footer_block_presence( automated_email.html_content ) if automated_email.is_opt_in: self.email_validator.validate_opt_in_button_block_presence( automated_email.html_content ) automated_email_copy = AutomatedEmail( name=request.name, type=automated_email.type, vendor_id=automated_email.vendor_id, subaccount_id=automated_email.subaccount_id, global_participant_id=automated_email.global_participant_id, custom_list_id=automated_email.custom_list_id, created_by=request.identity_id, updated_by=request.identity_id, ) automated_email_copy.sender_name = automated_email.sender_name automated_email_copy.email_username = automated_email.email_username automated_email_copy.email_domain = automated_email.email_domain automated_email_copy.email_domain_id = automated_email.email_domain_id automated_email_copy.subject = automated_email.subject automated_email_copy.html_content = automated_email.html_content automated_email_copy.css_content = automated_email.css_content automated_email_copy.custom_fonts = automated_email.custom_fonts automated_email_copy.preview_text = automated_email.preview_text automated_email_copy.source_campaign_connections = [] # Save the email campaign self.automated_email_repository.save(automated_email_copy) return automated_email_copy