"""Models for Campaign Approval flow.""" from datetime import date from typing import List from sqlalchemy import func, and_ from sqlalchemy.dialects import postgresql from db import db class ApprovalRequest(db.Model): """Approval Request model""" __tablename__ = "ApprovalRequest" id = db.Column(db.BigInteger, primary_key=True) email_uuid = db.Column(postgresql.UUID(as_uuid=True), unique=True, nullable=True) media_plan_id = db.Column(db.BigInteger, db.ForeignKey("MediaPlan.id", ondelete="CASCADE")) requested_at = db.Column(db.TIMESTAMP(timezone=True), nullable=False, server_default=func.current_timestamp()) requester_id = db.Column(db.Integer, db.ForeignKey("User.id"), nullable=False) reviewer_id = db.Column(db.Integer, db.ForeignKey("User.id"), nullable=False, index=True) message = db.Column(db.String, nullable=True) media_plan = db.relationship("MediaPlan", back_populates="approval_requests") requester = db.relationship("User", foreign_keys=[requester_id]) reviewer = db.relationship("User", foreign_keys=[reviewer_id]) approval_request_campaigns: List["ApprovalRequestCampaign"] = db.relationship( "ApprovalRequestCampaign", back_populates="approval_request", cascade="all, delete-orphan" ) def __str__(self): return f"Approval request id {self.id}" class ApprovalRequestCampaign(db.Model): """Campaign Approval Request model""" __tablename__ = "ApprovalRequestCampaign" id = db.Column(db.BigInteger, primary_key=True) approval_request_id = db.Column( db.BigInteger, db.ForeignKey("ApprovalRequest.id", ondelete="CASCADE"), nullable=False ) campaign_uuid = db.Column( postgresql.UUID(as_uuid=True), db.ForeignKey("Campaign.uuid", ondelete="CASCADE"), nullable=False, index=True ) expiration_date = db.Column(db.Date, nullable=False) completed_at = db.Column(db.TIMESTAMP(timezone=True), nullable=True) is_approved = db.Column(db.Boolean, nullable=False, server_default="FALSE") is_deleted = db.Column(db.Boolean, nullable=False, server_default="FALSE") approval_request = db.relationship("ApprovalRequest", foreign_keys=[approval_request_id]) campaign = db.relationship("Campaign", foreign_keys=[campaign_uuid]) __table_args__ = ( db.Index("unique_active_approval_request_campaign", campaign_uuid, unique=True, postgresql_where=and_(completed_at.is_(None), is_deleted == False) ), ) def __str__(self): return f"Approval request campaign id {self.id}" @property def is_live(self) -> bool: return self.completed_at is None and self.is_deleted is False and self.expiration_date > date.today()