| 1 |
8440155 |
mahdim10 |
1y ago |
from abc import ABC, abstractmethod |
| 2 |
8440155 |
mahdim10 |
1y ago |
from enum import Enum |
| 3 |
8440155 |
mahdim10 |
1y ago |
from functools import lru_cache |
| 4 |
8440155 |
mahdim10 |
1y ago |
from typing import Any, Dict, Type, Optional, Union |
| 5 |
8440155 |
mahdim10 |
1y ago |
|
| 6 |
8440155 |
mahdim10 |
1y ago |
from app.models.admin import Admin |
| 7 |
8440155 |
mahdim10 |
1y ago |
from app.models.notification import ( |
| 8 |
8440155 |
mahdim10 |
1y ago |
Notification, |
| 9 |
8440155 |
mahdim10 |
1y ago |
AdminNotif, |
| 10 |
8440155 |
mahdim10 |
1y ago |
UserNotification, |
| 11 |
8440155 |
mahdim10 |
1y ago |
UserCreated, |
| 12 |
8440155 |
mahdim10 |
1y ago |
UserUpdated, |
| 13 |
8440155 |
mahdim10 |
1y ago |
UserActivated, |
| 14 |
8440155 |
mahdim10 |
1y ago |
UserDeactivated, |
| 15 |
8440155 |
mahdim10 |
1y ago |
UserDeleted, |
| 16 |
8440155 |
mahdim10 |
1y ago |
UserEnabled, |
| 17 |
8440155 |
mahdim10 |
1y ago |
UserDisabled, |
| 18 |
8440155 |
mahdim10 |
1y ago |
UserDataUsageReset, |
| 19 |
8440155 |
mahdim10 |
1y ago |
UserSubscriptionRevoked, |
| 20 |
8440155 |
mahdim10 |
1y ago |
ReachedUsagePercent, |
| 21 |
8440155 |
mahdim10 |
1y ago |
ReachedDaysLeft, |
| 22 |
8440155 |
mahdim10 |
1y ago |
) |
| 23 |
8440155 |
mahdim10 |
1y ago |
from app.models.user import UserResponse |
| 24 |
8440155 |
mahdim10 |
1y ago |
|
| 25 |
8440155 |
mahdim10 |
1y ago |
|
| 26 |
8440155 |
mahdim10 |
1y ago |
class NotificationFactory(ABC): |
| 27 |
8440155 |
mahdim10 |
1y ago |
@abstractmethod |
| 28 |
8440155 |
mahdim10 |
1y ago |
def create_notification(self, action: Enum, **kwargs: Any) -> Notification: |
| 29 |
8440155 |
mahdim10 |
1y ago |
raise NotImplementedError() |
| 30 |
8440155 |
mahdim10 |
1y ago |
|
| 31 |
8440155 |
mahdim10 |
1y ago |
|
| 32 |
8440155 |
mahdim10 |
1y ago |
class AdminNotificationFactory(NotificationFactory): |
| 33 |
8440155 |
mahdim10 |
1y ago |
def create_notification(self, action: Enum, **kwargs: Any) -> Notification: |
| 34 |
8440155 |
mahdim10 |
1y ago |
pass |
| 35 |
8440155 |
mahdim10 |
1y ago |
|
| 36 |
8440155 |
mahdim10 |
1y ago |
|
| 37 |
8440155 |
mahdim10 |
1y ago |
class UserNotificationFactory(NotificationFactory): |
| 38 |
8440155 |
mahdim10 |
1y ago |
Action = UserNotification.Action |
| 39 |
8440155 |
mahdim10 |
1y ago |
|
| 40 |
8440155 |
mahdim10 |
1y ago |
notification_classes: Dict[ |
| 41 |
8440155 |
mahdim10 |
1y ago |
UserNotification.Action, Type[UserNotification] |
| 42 |
8440155 |
mahdim10 |
1y ago |
] = { |
| 43 |
8440155 |
mahdim10 |
1y ago |
Action.user_created: UserCreated, |
| 44 |
8440155 |
mahdim10 |
1y ago |
Action.user_updated: UserUpdated, |
| 45 |
8440155 |
mahdim10 |
1y ago |
Action.user_activated: UserActivated, |
| 46 |
8440155 |
mahdim10 |
1y ago |
Action.user_deactivated: UserDeactivated, |
| 47 |
8440155 |
mahdim10 |
1y ago |
Action.user_deleted: UserDeleted, |
| 48 |
8440155 |
mahdim10 |
1y ago |
Action.user_enabled: UserEnabled, |
| 49 |
8440155 |
mahdim10 |
1y ago |
Action.user_disabled: UserDisabled, |
| 50 |
8440155 |
mahdim10 |
1y ago |
Action.data_usage_reset: UserDataUsageReset, |
| 51 |
8440155 |
mahdim10 |
1y ago |
Action.subscription_revoked: UserSubscriptionRevoked, |
| 52 |
8440155 |
mahdim10 |
1y ago |
Action.reached_usage_percent: ReachedUsagePercent, |
| 53 |
8440155 |
mahdim10 |
1y ago |
Action.reached_days_left: ReachedDaysLeft, |
| 54 |
8440155 |
mahdim10 |
1y ago |
} |
| 55 |
8440155 |
mahdim10 |
1y ago |
|
| 56 |
8440155 |
mahdim10 |
1y ago |
def create_notification( |
| 57 |
8440155 |
mahdim10 |
1y ago |
self, |
| 58 |
8440155 |
mahdim10 |
1y ago |
action: UserNotification.Action, |
| 59 |
8440155 |
mahdim10 |
1y ago |
user: UserResponse, |
| 60 |
8440155 |
mahdim10 |
1y ago |
by: Optional[Admin] = None, |
| 61 |
8440155 |
mahdim10 |
1y ago |
**kwargs: Any |
| 62 |
8440155 |
mahdim10 |
1y ago |
) -> UserNotification: |
| 63 |
8440155 |
mahdim10 |
1y ago |
notification_class = self.notification_classes.get(action) |
| 64 |
8440155 |
mahdim10 |
1y ago |
return notification_class(user=user, by=by, **kwargs) |
| 65 |
8440155 |
mahdim10 |
1y ago |
|
| 66 |
8440155 |
mahdim10 |
1y ago |
|
| 67 |
8440155 |
mahdim10 |
1y ago |
class NotificationStrategy: |
| 68 |
8440155 |
mahdim10 |
1y ago |
def __init__(self): |
| 69 |
8440155 |
mahdim10 |
1y ago |
self.user_factory = UserNotificationFactory() |
| 70 |
8440155 |
mahdim10 |
1y ago |
self.admin_factory = AdminNotificationFactory() |
| 71 |
8440155 |
mahdim10 |
1y ago |
|
| 72 |
8440155 |
mahdim10 |
1y ago |
def create_notification( |
| 73 |
8440155 |
mahdim10 |
1y ago |
self, action: Enum, **kwargs: Any |
| 74 |
8440155 |
mahdim10 |
1y ago |
) -> Union[UserNotification, AdminNotif]: |
| 75 |
8440155 |
mahdim10 |
1y ago |
if isinstance(action, UserNotification.Action): |
| 76 |
8440155 |
mahdim10 |
1y ago |
return self.user_factory.create_notification(action, **kwargs) |
| 77 |
8440155 |
mahdim10 |
1y ago |
elif isinstance(action, AdminNotif.Action): |
| 78 |
8440155 |
mahdim10 |
1y ago |
return self.admin_factory.create_notification(action, **kwargs) |
| 79 |
8440155 |
mahdim10 |
1y ago |
|
| 80 |
8440155 |
mahdim10 |
1y ago |
|
| 81 |
8440155 |
mahdim10 |
1y ago |
@lru_cache(maxsize=1) |
| 82 |
8440155 |
mahdim10 |
1y ago |
def get_notification_strategy() -> NotificationStrategy: |
| 83 |
8440155 |
mahdim10 |
1y ago |
return NotificationStrategy() |