from __future__ import annotations from datetime import datetime import sqlalchemy as sa from fansifter_common.adapters.db.types import ChoiceType from fansifter_common.email import EmailType from fansifter_common.utils import timezone from fansifter_common.utils.uuid import uuid_string from sqlalchemy.orm import Mapped, mapped_column, query_expression, relationship from email_campaigns.adapters.db import Model from email_campaigns.automated.enums import AutomatedEmailType, SourceCampaignStatus from email_campaigns.emails.models import BaseEmail from email_campaigns.rosters.enums import FanDataListType from email_campaigns.rosters.types import FanDataListId class AutomatedEmail(BaseEmail, kw_only=True, repr=False): __tablename__ = "email_automated" email_type = EmailType.AUTOMATED id: Mapped[str] = mapped_column(primary_key=True, default_factory=uuid_string) type: Mapped[AutomatedEmailType] = mapped_column(ChoiceType(AutomatedEmailType)) last_sent_at: Mapped[datetime | None] = mapped_column(default=None) source_campaign_connections: Mapped[list[AutomatedEmailTrigger]] = relationship( lazy="selectin", cascade="all, delete-orphan", default_factory=list, repr=False, ) @property def is_automated(self) -> bool: return bool(self.source_campaign_connections) @property def is_opt_in(self) -> bool: return self.type == AutomatedEmailType.OPT_IN class SourceCampaign(Model, kw_only=True, repr=False): __tablename__ = "v_source_campaign" id: Mapped[str] = mapped_column(primary_key=True) vendor_id: Mapped[int | None] = mapped_column(default=None) subaccount_id: Mapped[int | None] = mapped_column(default=None) campaign_id: Mapped[str] source: Mapped[str] name: Mapped[str] url: Mapped[str | None] = mapped_column(default=None) status: Mapped[SourceCampaignStatus] = mapped_column( ChoiceType(SourceCampaignStatus), default=SourceCampaignStatus.DRAFT ) global_participant_id: Mapped[str | None] = mapped_column(default=None) custom_list_id: Mapped[str | None] = mapped_column(default=None) is_double_opt_in_enabled: Mapped[bool] = mapped_column(default=False) created_at: Mapped[datetime] = mapped_column(default_factory=timezone.now) updated_at: Mapped[datetime] = mapped_column(default_factory=timezone.now) has_other_connections: Mapped[bool | None] = query_expression() @property def fandata_list_id(self): if self.global_participant_id: return FanDataListId( id=self.global_participant_id, type=FanDataListType.ARTIST ) elif self.custom_list_id: return FanDataListId( id=self.custom_list_id, type=FanDataListType.CUSTOM_LIST ) return None class AutomatedEmailTrigger(Model, kw_only=True, repr=False): __tablename__ = "email_automated_trigger" automated_email_id: Mapped[str] = mapped_column( sa.ForeignKey(AutomatedEmail.id), primary_key=True, init=False, ) source_campaign_id: Mapped[str] = mapped_column( sa.ForeignKey(SourceCampaign.id), primary_key=True ) subscription_only: Mapped[bool] = mapped_column(default=False) # Relations source_campaign: Mapped[SourceCampaign] = relationship( lazy="selectin", viewonly=True, init=False, repr=False, )