Spaces:
Sleeping
Sleeping
File size: 5,376 Bytes
456b2e2 ab3ba46 456b2e2 157a11c 456b2e2 157a11c 456b2e2 157a11c 456b2e2 ab3ba46 456b2e2 e90be6b 456b2e2 157a11c 456b2e2 157a11c 456b2e2 157a11c 456b2e2 e90be6b 456b2e2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 | """
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"<Notification(id={self.id}, user_id={self.user_id}, type={self.notification_type}, status={self.status})>"
@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
|