""" Notification Model - Polymorphic notification system """ from sqlalchemy import Column, String, Text, DateTime, Enum as SQLEnum, TIMESTAMP, Index, ForeignKey from sqlalchemy.dialects.postgresql import UUID, JSONB from sqlalchemy.orm import relationship from sqlalchemy.sql import func import uuid import enum from app.core.database import Base class NotificationChannel(str, enum.Enum): """Notification delivery channels""" EMAIL = "email" SMS = "sms" WHATSAPP = "whatsapp" IN_APP = "in_app" PUSH = "push" class NotificationStatus(str, enum.Enum): """Notification delivery status""" PENDING = "pending" SENT = "sent" DELIVERED = "delivered" FAILED = "failed" READ = "read" class Notification(Base): """ Polymorphic notification system tracking notifications sent to users. Supports multiple channels (email, SMS, WhatsApp, in-app, push) and links to any source entity (tickets, expenses, projects, etc.) Note: Does not inherit BaseModel because database schema has only created_at and deleted_at, not updated_at. """ __tablename__ = "notifications" # Primary Key id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) # Recipient user_id = Column(UUID(as_uuid=True), ForeignKey('users.id'), nullable=False, index=True) # Project Scoping (for filtering notifications by project) project_id = Column(UUID(as_uuid=True), ForeignKey('projects.id', ondelete='SET NULL'), nullable=True, index=True) # Source (polymorphic - what triggered this notification) source_type = Column(String, nullable=False, index=True) # 'ticket', 'project', 'expense', 'compensation', 'system' source_id = Column(UUID(as_uuid=True), nullable=True, index=True) # ID of the source entity # Notification Content title = Column(Text, nullable=False) message = Column(Text, nullable=False) notification_type = Column(String, nullable=True) # 'assignment', 'status_change', 'approval', 'reminder', 'alert' # Delivery channel = Column(SQLEnum(NotificationChannel, name='notification_channel', values_callable=lambda x: [e.value for e in x]), nullable=False, index=True) status = Column(SQLEnum(NotificationStatus, name='notification_status', values_callable=lambda x: [e.value for e in x]), nullable=False, default=NotificationStatus.PENDING, index=True) # Delivery Tracking sent_at = Column(TIMESTAMP(timezone=True), nullable=True) delivered_at = Column(TIMESTAMP(timezone=True), nullable=True) read_at = Column(TIMESTAMP(timezone=True), nullable=True) failed_at = Column(TIMESTAMP(timezone=True), nullable=True) failure_reason = Column(Text, nullable=True) # Metadata additional_metadata = Column(JSONB, default={}, nullable=False) # Additional data (e.g., action buttons, deep links) # Timestamps (matches database schema exactly) created_at = Column(TIMESTAMP(timezone=True), server_default=func.timezone('utc', func.now()), nullable=False) deleted_at = Column(TIMESTAMP(timezone=True), nullable=True) # Soft delete # Relationships user = relationship("User", foreign_keys=[user_id], backref="notifications") project = relationship("Project", foreign_keys=[project_id], backref="notifications") # Indexes defined in table args __table_args__ = ( Index('idx_notifications_user', 'user_id', 'status', 'created_at'), Index('idx_notifications_source', 'source_type', 'source_id'), Index('idx_notifications_status', 'status', 'channel', 'created_at'), Index('idx_notifications_unread', 'user_id', 'status', 'created_at', postgresql_where=(Column('status') != NotificationStatus.READ)), ) def __repr__(self): return f"" @property def is_read(self) -> bool: """Check if notification has been read""" return self.read_at is not None or self.status == NotificationStatus.READ @property def is_sent(self) -> bool: """Check if notification has been sent""" return self.sent_at is not None or self.status in [ NotificationStatus.SENT, NotificationStatus.DELIVERED, NotificationStatus.READ ] def mark_as_read(self): """Mark notification as read""" from datetime import datetime, timezone self.read_at = datetime.now(timezone.utc) self.status = NotificationStatus.READ def mark_as_sent(self): """Mark notification as sent""" from datetime import datetime, timezone self.sent_at = datetime.now(timezone.utc) self.status = NotificationStatus.SENT def mark_as_delivered(self): """Mark notification as delivered""" from datetime import datetime, timezone self.delivered_at = datetime.now(timezone.utc) self.status = NotificationStatus.DELIVERED def mark_as_failed(self, reason: str): """Mark notification as failed""" from datetime import datetime, timezone self.failed_at = datetime.now(timezone.utc) self.failure_reason = reason self.status = NotificationStatus.FAILED