11c1cc811d
email_sender_hub.py: extract email_sender_manager = EmailSenderManager() notification_hub.py: extract notification_manager = NotificationManager() template_message_hub.py: extract template_message_manager = TemplateMessageManager() template_message_hub.py: add more details in render_template function
36 lines
1.2 KiB
Python
36 lines
1.2 KiB
Python
from typing import Dict
|
|
from backend.business.notification_manager import NotificationManager
|
|
from backend.models.constants import NotificationChannel
|
|
|
|
|
|
class NotificationHub:
|
|
def __init__(self):
|
|
self.notification_manager = NotificationManager()
|
|
|
|
async def enqueue_notification(
|
|
self,
|
|
channels: list[str], # Accept multiple channels as a list of strings
|
|
receiver_id: str,
|
|
subject: str,
|
|
event: str,
|
|
properties: Dict,
|
|
):
|
|
# Convert string channels to NotificationChannel enums
|
|
notification_channels = []
|
|
for channel_str in channels:
|
|
channel_int = int(channel_str)
|
|
try:
|
|
notification_channels.append(NotificationChannel(channel_int))
|
|
except KeyError:
|
|
raise ValueError(f"Unsupported notification channel: {channel_int}")
|
|
|
|
# Initialize NotificationManager with sender_id
|
|
# Call the enqueue_notification method in NotificationManager
|
|
return await self.notification_manager.enqueue_notification(
|
|
channels=notification_channels,
|
|
receiver_id=receiver_id,
|
|
subject=subject,
|
|
event=event,
|
|
properties=properties,
|
|
)
|