| 1 |
8440155 |
mahdim10 |
1y ago |
from functools import lru_cache |
| 2 |
8440155 |
mahdim10 |
1y ago |
from typing import List |
| 3 |
8440155 |
mahdim10 |
1y ago |
from abc import ABC, abstractmethod |
| 4 |
8440155 |
mahdim10 |
1y ago |
|
| 5 |
8440155 |
mahdim10 |
1y ago |
from app.config.env import TELEGRAM_API_TOKEN, WEBHOOK_ADDRESS |
| 6 |
8440155 |
mahdim10 |
1y ago |
from app.models.notification import Notification |
| 7 |
8440155 |
mahdim10 |
1y ago |
from app.notification import telegram, webhook |
| 8 |
8440155 |
mahdim10 |
1y ago |
|
| 9 |
8440155 |
mahdim10 |
1y ago |
|
| 10 |
8440155 |
mahdim10 |
1y ago |
class BaseNotificationService(ABC): |
| 11 |
8440155 |
mahdim10 |
1y ago |
@abstractmethod |
| 12 |
8440155 |
mahdim10 |
1y ago |
async def send_notification(self, message: Notification): |
| 13 |
8440155 |
mahdim10 |
1y ago |
raise NotImplementedError() |
| 14 |
8440155 |
mahdim10 |
1y ago |
|
| 15 |
8440155 |
mahdim10 |
1y ago |
|
| 16 |
8440155 |
mahdim10 |
1y ago |
class WebhookNotificationService(BaseNotificationService): |
| 17 |
8440155 |
mahdim10 |
1y ago |
async def send_notification(self, message: Notification) -> None: |
| 18 |
8440155 |
mahdim10 |
1y ago |
await webhook.send_notification(message) |
| 19 |
8440155 |
mahdim10 |
1y ago |
|
| 20 |
8440155 |
mahdim10 |
1y ago |
|
| 21 |
8440155 |
mahdim10 |
1y ago |
class TelegramNotificationService(BaseNotificationService): |
| 22 |
8440155 |
mahdim10 |
1y ago |
async def send_notification(self, message: Notification) -> None: |
| 23 |
8440155 |
mahdim10 |
1y ago |
await telegram.send_notification(message) |
| 24 |
8440155 |
mahdim10 |
1y ago |
|
| 25 |
8440155 |
mahdim10 |
1y ago |
|
| 26 |
8440155 |
mahdim10 |
1y ago |
class NotificationManager: |
| 27 |
8440155 |
mahdim10 |
1y ago |
def __init__(self, strategies: List[BaseNotificationService]) -> None: |
| 28 |
8440155 |
mahdim10 |
1y ago |
self.strategies = strategies |
| 29 |
8440155 |
mahdim10 |
1y ago |
|
| 30 |
8440155 |
mahdim10 |
1y ago |
async def send_notification(self, message: Notification) -> None: |
| 31 |
8440155 |
mahdim10 |
1y ago |
for strategy in self.strategies: |
| 32 |
8440155 |
mahdim10 |
1y ago |
await strategy.send_notification(message) |
| 33 |
8440155 |
mahdim10 |
1y ago |
|
| 34 |
8440155 |
mahdim10 |
1y ago |
|
| 35 |
8440155 |
mahdim10 |
1y ago |
@lru_cache(maxsize=1) |
| 36 |
8440155 |
mahdim10 |
1y ago |
def get_notification_manager() -> NotificationManager: |
| 37 |
8440155 |
mahdim10 |
1y ago |
services = [] |
| 38 |
8440155 |
mahdim10 |
1y ago |
|
| 39 |
8440155 |
mahdim10 |
1y ago |
if TELEGRAM_API_TOKEN: |
| 40 |
8440155 |
mahdim10 |
1y ago |
services.append(TelegramNotificationService()) |
| 41 |
8440155 |
mahdim10 |
1y ago |
|
| 42 |
8440155 |
mahdim10 |
1y ago |
if WEBHOOK_ADDRESS: |
| 43 |
8440155 |
mahdim10 |
1y ago |
services.append(WebhookNotificationService()) |
| 44 |
8440155 |
mahdim10 |
1y ago |
|
| 45 |
8440155 |
mahdim10 |
1y ago |
if not services: |
| 46 |
8440155 |
mahdim10 |
1y ago |
raise ValueError("No notification services provided.") |
| 47 |
8440155 |
mahdim10 |
1y ago |
|
| 48 |
8440155 |
mahdim10 |
1y ago |
return NotificationManager(services) |