Coverage for src / lexigram / admin / services / notifications / models.py: 96%

1"""Notification data models and types for admin notifications.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from datetime import UTC, datetime 

7from enum import StrEnum 

8from typing import Any 

9 

10 

11class NotificationType(StrEnum): 

12 """Types of admin notifications.""" 

13 

14 # User events 

15 USER_CREATED = "user_created" 

16 USER_INVITED = "user_invited" 

17 USER_ACTIVATED = "user_activated" 

18 USER_DEACTIVATED = "user_deactivated" 

19 PASSWORD_RESET = "password_reset" 

20 PASSWORD_CHANGED = "password_changed" 

21 

22 # Bulk operations 

23 BULK_STARTED = "bulk_started" 

24 BULK_PROGRESS = "bulk_progress" 

25 BULK_COMPLETED = "bulk_completed" 

26 BULK_FAILED = "bulk_failed" 

27 

28 # System events 

29 SYSTEM_ALERT = "system_alert" 

30 SYSTEM_WARNING = "system_warning" 

31 SYSTEM_ERROR = "system_error" 

32 

33 # Export/Import 

34 EXPORT_READY = "export_ready" 

35 IMPORT_COMPLETED = "import_completed" 

36 IMPORT_FAILED = "import_failed" 

37 

38 

39class NotificationChannel(StrEnum): 

40 """Notification delivery channels.""" 

41 

42 EMAIL = "email" 

43 IN_APP = "in_app" 

44 PUSH = "push" 

45 

46 

47@dataclass 

48class NotificationRecipient: 

49 """Notification recipient.""" 

50 

51 email: str 

52 name: str | None = None 

53 user_id: Any = None 

54 preferences: dict[str, bool] = field(default_factory=dict) 

55 

56 def can_receive(self, notification_type: NotificationType) -> bool: 

57 """Check if recipient can receive notification type.""" 

58 key = f"notify_{notification_type.value}" 

59 return self.preferences.get(key, True) 

60 

61 

62@dataclass 

63class Notification: 

64 """Admin notification.""" 

65 

66 type: NotificationType 

67 subject: str 

68 body: str 

69 recipients: list[NotificationRecipient] 

70 channels: list[NotificationChannel] = field( 

71 default_factory=lambda: [NotificationChannel.EMAIL], 

72 ) 

73 

74 # Optional fields 

75 data: dict[str, Any] = field(default_factory=dict) 

76 html_body: str | None = None 

77 priority: str = "normal" 

78 scheduled_at: datetime | None = None 

79 

80 # Tracking 

81 id: str | None = None 

82 created_at: datetime = field(default_factory=lambda: datetime.now(UTC)) 

83 

84 

85@dataclass 

86class NotificationResult: 

87 """Payload of a completed notification delivery. 

88 

89 Always carried inside ``Ok[NotificationResult, NotificationError]``. 

90 Per-recipient failures are tracked in ``errors`` and ``recipients_failed``; 

91 a non-zero ``recipients_failed`` does NOT mean the overall operation failed — 

92 only an ``Err`` return (when every recipient failed) does. 

93 """ 

94 

95 notification_id: str | None = None 

96 recipients_sent: int = 0 

97 recipients_failed: int = 0 

98 errors: list[str] = field(default_factory=list) 

99 

100 

101__all__ = [ 

102 "Notification", 

103 "NotificationChannel", 

104 "NotificationRecipient", 

105 "NotificationResult", 

106 "NotificationType", 

107]