from datetime import datetime import sqlalchemy as sa from fansifter_common.utils import timezone from fansifter_common.utils.uuid import uuid_string from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy from sqlalchemy.orm import Mapped, mapped_column, relationship from email_campaigns.adapters.db import Model class EmailReportingReportEmail(Model, kw_only=True): __tablename__ = "email_reporting_report_email" report_id: Mapped[str] = mapped_column( sa.ForeignKey("email_reporting_report.id", ondelete="CASCADE"), primary_key=True, init=False, ) email_id: Mapped[str] = mapped_column(primary_key=True) class EmailReportingReport(Model, kw_only=True): __tablename__ = "email_reporting_report" __table_args__ = (sa.UniqueConstraint("identity_id", "name"),) id: Mapped[str] = mapped_column(primary_key=True, default_factory=uuid_string) identity_id: Mapped[str] name: Mapped[str] created_at: Mapped[datetime] = mapped_column(default_factory=timezone.now) updated_at: Mapped[datetime] = mapped_column( default_factory=timezone.now, onupdate=timezone.now ) report_emails: Mapped[list[EmailReportingReportEmail]] = relationship( default_factory=list, cascade="all, delete-orphan", passive_deletes=True, lazy="selectin", init=False, repr=False, ) _email_ids: AssociationProxy[list[str]] = association_proxy( "report_emails", "email_id", default_factory=list, creator=lambda email_id: EmailReportingReportEmail(email_id=email_id), ) @property def email_ids(self) -> list[str]: return list(self._email_ids) @email_ids.setter def email_ids(self, email_ids: list[str]) -> None: self._email_ids = email_ids @property def emails_count(self) -> int: return len(self.report_emails)