38 lines
1.1 KiB
Python
38 lines
1.1 KiB
Python
import httpx
|
|
from common.config.app_settings import app_settings
|
|
from typing import Dict, List
|
|
|
|
|
|
class NotificationService:
|
|
def __init__(self):
|
|
self.notification_api_url = app_settings.NOTIFICATION_WEBAPI_URL_BASE.rstrip(
|
|
"/"
|
|
)
|
|
|
|
async def send_notification(
|
|
self,
|
|
sender_id: str,
|
|
channels: List[str],
|
|
receiver_id: str,
|
|
subject: str,
|
|
event: str,
|
|
properties: Dict,
|
|
) -> bool:
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.post(
|
|
f"{self.notification_api_url}/send_notification",
|
|
json={
|
|
"sender_id": sender_id,
|
|
"channels": channels,
|
|
"receiver_id": receiver_id,
|
|
"subject": subject,
|
|
"event": event,
|
|
"properties": properties,
|
|
},
|
|
)
|
|
if response.status_code == 200:
|
|
return True
|
|
else:
|
|
# Optionally log or handle errors here
|
|
return False
|