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

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" # noqa: S105 # event type name, not a credential 

20 EMAIL_VERIFICATION = "email_verification" 

21 EMAIL_OTP = "email_otp" 

22 PASSWORD_CHANGED = "password_changed" # noqa: S105 # event type name, not a credential 

23 

24 # Bulk operations 

25 BULK_STARTED = "bulk_started" 

26 BULK_PROGRESS = "bulk_progress" 

27 BULK_COMPLETED = "bulk_completed" 

28 BULK_FAILED = "bulk_failed" 

29 

30 # System events 

31 SYSTEM_ALERT = "system_alert" 

32 SYSTEM_WARNING = "system_warning" 

33 SYSTEM_ERROR = "system_error" 

34 

35 # Export/Import 

36 EXPORT_READY = "export_ready" 

37 IMPORT_COMPLETED = "import_completed" 

38 IMPORT_FAILED = "import_failed" 

39 

40 

41class NotificationChannel(StrEnum): 

42 """Notification delivery channels.""" 

43 

44 EMAIL = "email" 

45 IN_APP = "in_app" 

46 PUSH = "push" 

47 

48 

49@dataclass 

50class NotificationRecipient: 

51 """Notification recipient.""" 

52 

53 email: str 

54 name: str | None = None 

55 user_id: Any = None 

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

57 

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

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

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

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

62 

63 

64@dataclass 

65class Notification: 

66 """Admin notification.""" 

67 

68 type: NotificationType 

69 subject: str 

70 body: str 

71 recipients: list[NotificationRecipient] 

72 channels: list[NotificationChannel] = field( 

73 default_factory=lambda: [NotificationChannel.EMAIL], 

74 ) 

75 

76 # Optional fields 

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

78 html_body: str | None = None 

79 priority: str = "normal" 

80 scheduled_at: datetime | None = None 

81 

82 # Tracking 

83 id: str | None = None 

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

85 

86 

87@dataclass 

88class NotificationResult: 

89 """Payload of a completed notification delivery. 

90 

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

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

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

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

95 """ 

96 

97 notification_id: str | None = None 

98 recipients_sent: int = 0 

99 recipients_failed: int = 0 

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

101 

102 

103__all__ = [ 

104 "Notification", 

105 "NotificationChannel", 

106 "NotificationRecipient", 

107 "NotificationResult", 

108 "NotificationType", 

109]