from dataclasses import dataclass from anydi import singleton from fansifter_common.auth.requests import AuthAccountRequest from fansifter_common.auth.services import AuthService from fansifter_common.auth.types import Permission from fansifter_common.auth.validators import AccountValidator from email_campaigns.adapters.db.decorators import transactional from email_campaigns.campaigns.models import EmailCampaign from email_campaigns.campaigns.repositories import EmailCampaignRepository from email_campaigns.campaigns.validators import CampaignValidator from email_campaigns.emails.services import EmailDomainService from email_campaigns.rosters.services import FanDataListService from email_campaigns.rosters.types import FanDataListId @dataclass(kw_only=True) class CreateCampaignRequest(AuthAccountRequest): name: str fandata_list_id: FanDataListId html_content: str | None = None css_content: str | None = None @singleton class CreateCampaignHandler: """Handler to create email campaigns.""" permission = Permission("email_campaign", "create") def __init__( self, auth_service: AuthService, account_validator: AccountValidator, email_domain_service: EmailDomainService, fandata_list_service: FanDataListService, campaign_repository: EmailCampaignRepository, campaign_validator: CampaignValidator, ) -> None: self.auth_service = auth_service self.account_validator = account_validator self.email_domain_service = email_domain_service self.fandata_list_service = fandata_list_service self.campaign_repository = campaign_repository self.campaign_validator = campaign_validator @transactional def handle(self, request: CreateCampaignRequest) -> EmailCampaign: # Check user permissions self.auth_service.authorize_account( request.identity_id, account=request.account, permission=self.permission, ) # Validate account self.account_validator.validate_account(request.account, allow_subaccount=False) # Validate campaign name uniqueness self.campaign_validator.validate_name(request.name, account=request.account) # Ensure account has access to artist fandata list self.fandata_list_service.ensure_artist_access( request.fandata_list_id, account=request.account ) # Resolve fandata list and set sender name sender_name = "" if request.fandata_list_id.is_artist: global_participant = self.fandata_list_service.resolve_global_participant( request.fandata_list_id.id ) sender_name = global_participant.name elif request.fandata_list_id.is_custom_list: custom_list = self.fandata_list_service.resolve_allowed_custom_list( request.fandata_list_id.id, account=request.account ) sender_name = custom_list.name # Setup email domain and username email_domain = self.email_domain_service.get_allowed_domain( identity_id=request.identity_id, account=request.account ) email_username = None if email_domain and sender_name: email_username = self.email_domain_service.create_email_username( sender_name=sender_name, domain_name=email_domain.domain ) # Create the campaign email = EmailCampaign( name=request.name, vendor_id=request.account.vendor_id, subaccount_id=request.account.subaccount_id, created_by=request.identity_id, updated_by=request.identity_id, ) email.sender_name = sender_name email.email_username = email_username email.email_domain = email_domain email.fandata_list_id = request.fandata_list_id email.css_content = request.css_content email.html_content = request.html_content # Save campaign self.campaign_repository.save(email) return email