from redis_wrapper.client import RedisCache


class UserCache(RedisCache):
    _key_prefix = "user"
    expire_time = 60 * 60

    @classmethod
    async def set_user(cls, key, value):
        return await cls.set(key, value, cls.expire_time)

app/caches/user.py
tata1mg/notifyone-gateway
__all__ = [
    "ErrorMessages",
    "NotificationChannels",
    "EventPriority",
    "PrepareNotification",
    "NotificationRequestLogStatus",
    "ProcessingType"
]

from .error_messages import ErrorMessages
from .notification_channels import NotificationChannels
from .event_priority import EventPriority
from .constants import PrepareNotification
from .notification_request_log_status import NotificationRequestLogStatus
from .process_type import ProcessingType
app/constants/__init__.py
tata1mg/notifyone-gateway
class PrepareNotification:
    ENDPOINT = "/v4/prepare-notification"
    METHOD = "POST"

app/constants/constants.py
tata1mg/notifyone-gateway
from commonutils.utils import CustomEnum


class ErrorMessages(CustomEnum):
    MISSING_REQUIRED_PARAMS = "missing required parameter - {}"
    INVALID_EMAIL_FORMAT = "invalid email format"
    INVALID_MOBILE_FORMAT = "invalid mobile format"
    EMAIL_MANDATORY_FOR_EVENT = 'email is mandatory to send notificatios for this event'
    MOBILE_MANDATORY_FOR_EVENT = 'mobile is mandatory to send notificatios for this event'
    INVALID_EVENT_PRIORITY = 'invalid event priority'
    EVENT_NOT_CONFIGURED = "Event not configured, configure it first. If you configured a new event recently, wait for few minnutes before you start sending notifications for that event"
    NO_CHANNEL_ACTIVE = "No notification channels are active for this event. Please enable at least one channel to send notifications"
    ONLY_ONE_TO_EMAIL_SUPPORTED = "Currently, only one email address is supported in 'to' address"
    ONLY_ONE_TO_MOBILE_SUPPORTED = "Currently, only one mobile number is supported in 'to' address"
    ONLY_ONE_TO_DEVICE_SUPPORTED = "Currently, only one device is supported in 'to' address for this push request"
    INVALID_ATTACHMENT_DATA = "Invalid attachment data - attachments must be a list of objects with each object containing exactly two properties - 'url' and 'filename'"

app/constants/error_messages.py
tata1mg/notifyone-gateway
from commonutils.utils import CustomEnum


class EventPriority(CustomEnum):
    CRITICAL = 'critical'
    HIGH = 'high'
    MEDIUM = 'medium'
    LOW = 'low'

    @classmethod
    def _missing_(cls, _value):
        return cls.LOW
app/constants/event_priority.py
tata1mg/notifyone-gateway
from commonutils.utils import CustomEnum


class NotificationChannels(CustomEnum):
    EMAIL = 'email'
    SMS = 'sms'
    WHATSAPP = 'whatsapp'
    PUSH = 'push'
    CALL = 'call'

    @classmethod
    def email_mandatory_channels(cls):
        return [cls.EMAIL.value]

    @classmethod
    def mobile_mandatory_channels(cls):
        return [cls.SMS.value, cls.WHATSAPP.value, cls.CALL.value]

app/constants/notification_channels.py
tata1mg/notifyone-gateway
from commonutils.utils import CustomEnum


class NotificationRequestLogStatus(str, CustomEnum):
    NEW = "NEW"
    INITIATED = "INITIATED"
    FAILED = "FAILED"
    SUCCESS = "SUCCESS"
    USER_OPT_OUT = "USER_OPT_OUT"
    NOT_ELIGIBLE = "NOT_ELIGIBLE"

    def __str__(self):
        return self.value

app/constants/notification_request_log_status.py
tata1mg/notifyone-gateway
from commonutils.utils import CustomEnum


class ProcessingType(str, CustomEnum):
    SYNC = "SYNC"
    ASYNC = "ASYNC"
    UNKNOWN = "UNKNOWN"

    def __str__(self):
        return self.value

app/constants/process_type.py
tata1mg/notifyone-gateway
__all__ = ["listeners"]

from .sync_app_events import sync_app_events_listeners
from .listener import other_listeners

listeners = sync_app_events_listeners + other_listeners

app/listeners/__init__.py
tata1mg/notifyone-gateway
import asyncio

from torpedo.constants import ListenerEventTypes

from app.services import EventNotification


async def setup_dispatch_notification(app, loop):
    EventNotification.setup()


other_listeners = [
    (setup_dispatch_notification, ListenerEventTypes.AFTER_SERVER_START.value)
]

app/listeners/listener.py
tata1mg/notifyone-gateway
import asyncio

from torpedo.constants import ListenerEventTypes

from app.services import Notifications


async def load_app_events(app, loop):
    asyncio.create_task(Notifications.app_events())


async def setup_app_events_periodic_refresh(app, loop):
    # Refresh app events mapping periodically
    asyncio.create_task(Notifications.refresh_app_events_periodically())

sync_app_events_listeners = [
    (setup_app_events_periodic_refresh, ListenerEventTypes.AFTER_SERVER_START.value),
    (load_app_events, ListenerEventTypes.AFTER_SERVER_START.value)
]

app/listeners/sync_app_events.py
tata1mg/notifyone-gateway
from app.constants import NotificationChannels, EventPriority


class Event:

    def __init__(self, event_dict: dict):
        self.id = event_dict['id']
        self.app_name = event_dict['app_name']
        self.event_name = event_dict['event_name']
        self.actions = event_dict['actions'] or dict()
        self.meta_info = event_dict['meta_info'] or dict()
        self.priority = EventPriority(self.meta_info.get('priority'))

    def any_channel_active(self) -> bool:
        return bool(sum(self.actions.values()))

    def is_email_mandatory(self) -> bool:
        return bool(sum([self.actions.get(channel, 0) for channel in NotificationChannels.email_mandatory_channels()]))

    def is_mobile_mandatory(self) -> bool:
        return bool(sum([self.actions.get(channel, 0) for channel in NotificationChannels.mobile_mandatory_channels()]))

app/models/event.py
tata1mg/notifyone-gateway
from sanic import Blueprint

from app.routes.notifications.notification import event_notification_blueprint

blueprint_group = Blueprint.group(
    event_notification_blueprint
)
app/routes/__init__.py
tata1mg/notifyone-gateway
from sanic_openapi import openapi


class BaseApiModel:

    _uri = ""
    _name = ""
    _method: str = "GET"
    _summary = ""
    _description = ""
    _request_content_type = "application/json"
    _response_content_type = "application/json"

    @classmethod
    def uri(cls):
        return cls._uri

    @classmethod
    def name(cls):
        return cls._name

    @classmethod
    def http_method(cls):
        return cls._method

    @classmethod
    def summary(cls):
        return cls._summary

    @classmethod
    def description(cls):
        return cls._description

    @classmethod
    def request_content_type(cls):
        return cls._request_content_type

    @classmethod
    def response_content_type(cls):
        return cls._response_content_type

    class RequestBodyOpenApiModel:
        """
        Represents the request body model
        """

    class RequestParamsOpenApiModel:
        """
        Represents the request query param model
        """

    class ResponseBodyOpenApiModel:
        """
        Represents the response body model
        """


class File:
    url = openapi.String(
        descritpion="File url",
        example="https://1mg-odin-production.s3.ap-south-1.amazonaws.com/upload/sales_orders/42550349/6f55151e-adb5-4171-8fe2-5eb6599eafb7.pdf",
        required=True,
    )
    filename = openapi.String(
        descritpion="File name", example="report.pdf", required=True
    )

app/routes/base_api_model.py
tata1mg/notifyone-gateway
from torpedo.constants import HTTPMethod
from sanic_openapi import openapi

from app.routes.base_api_model import BaseApiModel, File


class ToOpenApiModel:
    email = openapi.Array(openapi.String(), desciption="List of email ids, required for email", required=False, example=["abc@gmail.com"])
    mobile = openapi.Array(openapi.String(), desciption="List of 10 digit mobile numbers, required for sms/whatsapp", required=False, example=["6000000000"])
    device = openapi.Array(openapi.String(), desciption="List of device registration ids (FCM), required for push notification", required=False, example=["device-id-1"])


class SenderOpenApiModel:
    name = openapi.String(description="Send name", example="Tata 1mg", required=False)
    address = openapi.String(description="Send email address", example="xyz@1mg.com", required=False)


class EmailChannelOpenApiModel:
    reply_to = openapi.String(description="Reply-to email ID", emample="reply-to@1mg.com", required=False)
    sender: SenderOpenApiModel

class ChannelsOpenApiModel:
    email = EmailChannelOpenApiModel

class NotificationOpenApiModel:
    id = openapi.Integer(example=3250, description="Notification Identifier")
    event_id = openapi.Integer(example=25, description="Event ID for this notification")
    notification_request_id = openapi.String(example="3e1ba047-71b1-4bd1-b510-613a50030b69", description="Unique ID generated by the Notification System")
    channel = openapi.String(example="push", description="Channel type of notification")
    status = openapi.String(example="SUCCESS", description="Current status. Possible values are - NEW/INITIATED/FAILED/SUCCESS/NOT_ELIGIBLE")
    sent_to = openapi.String(example="DEVICE1211212", description="Recipient address")
    source_identifier = openapi.String(example="ID1212122", description="Source identifier sent with the request")
    operator = openapi.String(example="FCM", description="Operator used to send out the notification")
    operator_event_id = openapi.String(example="A112123232323", description="Operator side ID for this notification")
    message = openapi.String(example="Success", description="Optional message")
    created = openapi.Integer(example=1679969481, description="Epoch time when the notification was created")
    updated = openapi.Integer(example=1679969481, description="Epoch time when the notification was last updated")


class GetNotificationApiResponseDataNotificationsOpenApiModel:
    notifications = openapi.Array(NotificationOpenApiModel)

class GetNotificationApiResponseDataOpenApiModel:
    data = GetNotificationApiResponseDataNotificationsOpenApiModel


class SendNotificationApiResponseDataOpenApiModel:
    request_id = openapi.String(description="Unique request ID generated by the Notification System for this request", example="3e1ba047-71b1-4bd1-b510-613a50030b69")
    message = openapi.String(description="a message string", example="success")
    processing_type = openapi.String(description="Processing type used for this request. It can SYNC or ASYNC", example="ASYNC")


class SendNotificationApiRequestBodyOrderOpenApiModel:
    order_id = openapi.String(example="PO21212121")

class SendNotificationApiRequestBodyOpenApiModel:
    order = SendNotificationApiRequestBodyOrderOpenApiModel


class GetNotificationApiModel(BaseApiModel):

    _uri = "/get-notification"
    _name = "get_notification"
    _method = HTTPMethod.GET.value
    _summary = "API to get status of the triggered notification"
    _description = (
        "This API can be used to fetch the status of a notification request."
        "It returns list of notifications triggered for the notification request"
    )

    class RequestParamsOpenApiModel:
        notification_request_id = openapi.String(description="Request ID retured by the system for a notification request", example="3e1ba047-71b1-4bd1-b510-613a50030b69", required=True)

    class ResponseBodyOpenApiModel:
        data = GetNotificationApiResponseDataOpenApiModel
        is_success = openapi.Boolean(description="true/false to denote if the request was successful or not", example=True)
        status_code = openapi.Integer(description="Http response status code", example=200)


class SendNotificationApiModel(BaseApiModel):

    _uri = "/send-notification"
    _name = "send_notification"
    _method = HTTPMethod.POST.value
    _summary = "API to sent out event notification"
    _description = (
        "This API can be used to trigger notifications for an already created event."
        "The API triggers notifications for all the active and eligible channels in the event. The processing can either be ASYNC or SYNC depending upon the event priority"
    )

    class RequestBodyOpenApiModel:
        event_id = openapi.Integer(example=1, description="Event ID for this notification")
        source_identifier = openapi.String(example="ID1212122", description="Source identifier sent with the request")
        to = ToOpenApiModel
        channels = ChannelsOpenApiModel
        attachments = openapi.Array(File, required=False, descriptions="List of files to be attached with the email/whatsapp")
        body = SendNotificationApiRequestBodyOpenApiModel

    class ResponseBodyOpenApiModel:
        data = SendNotificationApiResponseDataOpenApiModel
        is_success = openapi.Boolean(description="true/false to denote if the request was successful or not", example=True)
        status_code = openapi.Integer(description="Http response status code", example=200)

app/routes/notifications/api_model.py
tata1mg/notifyone-gateway
import logging

from sanic import Blueprint
from sanic_openapi import openapi
from sanic_openapi.openapi3.definitions import Parameter
from torpedo import Request, send_response
from torpedo.exceptions import BadRequestException

from app.services import EventNotification, Notifications
from app.routes.notifications.api_model import GetNotificationApiModel, SendNotificationApiModel
from app.constants import ErrorMessages

event_notification_blueprint = Blueprint("event_notification")

logger = logging.getLogger()


@event_notification_blueprint.route(
    SendNotificationApiModel.uri(), methods=[SendNotificationApiModel.http_method()], name=SendNotificationApiModel.name()
)
@openapi.definition(
    summary=SendNotificationApiModel.summary(),
    description=SendNotificationApiModel.description(),
    body={
        SendNotificationApiModel.request_content_type(): SendNotificationApiModel.RequestBodyOpenApiModel
    },
    response={
        SendNotificationApiModel.response_content_type(): SendNotificationApiModel.ResponseBodyOpenApiModel
    },
)
async def send_event_notification(request: Request):
    """
    Endpoint to trigger event based notifications
    """
    request_payload = request.custom_json()
    event_id = request_payload.get('event_id')
    source_identifier = request_payload.get('source_identifier')
    to_addresses = request_payload.get('to') or dict()
    if not event_id:
        raise BadRequestException(ErrorMessages.MISSING_REQUIRED_PARAMS.value.format('event_id'))
    if not source_identifier:
        raise BadRequestException(ErrorMessages.MISSING_REQUIRED_PARAMS.value.format('source_identifier'))

    try:
        data = await EventNotification.trigger_event_notification(request_payload)
    except Exception as e:
        logger.error(
            "Request Failed - error type : {}, message : {}, params - {} {}".format(
                type(e), str(e), event_id, to_addresses
            )
        )
        raise e
    return send_response(data=data)


@event_notification_blueprint.route(
    GetNotificationApiModel.uri(), methods=[GetNotificationApiModel.http_method()], name=GetNotificationApiModel.name()
)
@openapi.definition(
    summary=GetNotificationApiModel.summary(),
    description=GetNotificationApiModel.description(),
    parameter=[Parameter('notification_request_id', openapi.String(), description="Unique ID generated by the Notification System", required=False), Parameter('source_identifier', openapi.String(), description="Source identifier sent with the request", required=False)],
    response={
        GetNotificationApiModel.response_content_type(): GetNotificationApiModel.ResponseBodyOpenApiModel
    },
)
async def get_event_notification(request: Request):
    """
    Endpoint to trigger event based notifications
    """
    request_payload = request.request_params()
    notification_request_id = request_payload.get('notification_request_id')
    source_identifier = request_payload.get('source_identifier')
    if not (notification_request_id or source_identifier):
        raise BadRequestException("Either notification_request_id OR source_identifier must be provided")

    try:
        data = await Notifications.get_event_notification(notification_request_id)
    except Exception as e:
        logger.error(
            "Request Failed - error type : {}, message : {}, params - {}".format(
                type(e), str(e), notification_request_id
            )
        )
        raise e
    return send_response(data=data)

app/routes/notifications/notification.py
tata1mg/notifyone-gateway
from torpedo import Host
from app.listeners import listeners
from app.routes import blueprint_group

if __name__ == "__main__":

    # register combined blueprint group here.
    # these blueprints are defined in the routes directory and has to be
    # collected in init file otherwise route will end up with 404 error.
    Host._listeners = listeners
    Host._blueprint_group = blueprint_group
    Host.run()

app/service.py
tata1mg/notifyone-gateway
__all__ = ["NotificationServiceClient"]

from .notification_service import NotificationServiceClient

app/service_clients/__init__.py
tata1mg/notifyone-gateway
import aiotask_context as context
from torpedo import BaseApiRequest
from torpedo.constants import X_SHARED_CONTEXT


class APIClient(BaseApiRequest):
    """
    Base class for all inter service requests. Each method will return

    AsyncTaskResponse(self._data['data'],
                      meta=self._data.get('meta', None),
                      status_code=self._data['status_code'],
                      headers=headers)
    """

    _timeout = None

    @classmethod
    def get_inter_service_headers(
        cls, headers_keys: list = None, get_shared_context=False
    ):
        headers_keys = headers_keys or list()
        _headers = {}
        headers = {}
        if headers:
            if headers_keys:
                for _key in headers_keys:
                    _headers[_key] = headers.get(_key)
            else:
                _headers = headers
            if get_shared_context:
                if context.get(X_SHARED_CONTEXT):
                    _headers[X_SHARED_CONTEXT] = context.get(X_SHARED_CONTEXT)
        return _headers

app/service_clients/base_api_client.py
tata1mg/notifyone-gateway
__all__ = ["NotificationServiceClient"]

from .notification_service import NotificationServiceClient

app/service_clients/notification_service/__init__.py
tata1mg/notifyone-gateway
from torpedo import CONFIG

from app.service_clients.base_api_client import APIClient


class NotificationServiceClient(APIClient):
    """
    auth service client
    """
    ns_config = CONFIG.config['NOTIFICATION_SERVICE']
    _host = ns_config['HOST']
    _timeout = ns_config['TIMEOUT']

    @classmethod
    async def get_events_custom(cls, attributes: list=None):
        """
        By default, this routine returns just the id, event_name and app_name details. Add extra attributes needed
        in the event_other_details param
        :param attributes: list of event attributes needed in the response
        :return:
        """
        # TODO remove this mocking after dev test
        path = '/events/custom'
        attributes = attributes or list()
        attributes_str = ','.join(attributes)
        params = {
            'attributes': attributes_str
        }
        result = await cls.get(path, query_params=params)
        return result.data

    @classmethod
    async def get_sent_notifications_by_notification_request_id(cls, notification_request_id):
        path = '/notifications/{notification_request_id}'.format(notification_request_id=notification_request_id)
        result = await cls.get(path)
        return result.data

app/service_clients/notification_service/notification_service.py
tata1mg/notifyone-gateway
__all__ = ["Notifications", "EventNotification"]

from .notifications import Notifications
from .event_notification import EventNotification
app/services/__init__.py
tata1mg/notifyone-gateway
import logging
from typing import Dict

from torpedo import CONFIG
from torpedo.exceptions import BadRequestException

from .notifications import Notifications
from .publisher import Publisher

from app.constants import ErrorMessages, EventPriority, PrepareNotification
from app.models import Event
from app.utilities import generate_uuid
from app.utilities.pubsub import SQSWrapper
from app.utilities.http import RestApiClientWrapper
from app.utilities.validators import validate_email, validate_mobile

logger = logging.getLogger()

notification_config = CONFIG.config["TRIGGER_NOTIFICATIONS"]


class EventNotification:
    HANDLERS: Dict[EventPriority, Publisher]

    @classmethod
    def setup(cls):
        critical_p_config = notification_config["CRITICAL_PRIORITY"]
        critical_priority_handler = RestApiClientWrapper(
            host=critical_p_config.get("HOST"),
            endpoint=PrepareNotification.ENDPOINT,
            method=PrepareNotification.METHOD,
        )
        high_priority_handler = SQSWrapper(
            notification_config["HIGH_PRIORITY"]["QUEUE_NAME"],
            config=notification_config,
        )
        medium_priority_handler = SQSWrapper(
            notification_config["MEDIUM_PRIORITY"]["QUEUE_NAME"],
            config=notification_config,
        )
        low_priority_handler = SQSWrapper(
            notification_config["LOW_PRIORITY"]["QUEUE_NAME"],
            config=notification_config,
        )

        cls.HANDLERS = {
            EventPriority.CRITICAL: critical_priority_handler,
            EventPriority.HIGH: high_priority_handler,
            EventPriority.MEDIUM: medium_priority_handler,
            EventPriority.LOW: low_priority_handler,
        }

    @staticmethod
    def _validate_notification_request(event: Event, payload: dict):
        to_addresses = payload.get('to') or dict()
        email_addresses = to_addresses.get('email') or list()
        mobile_numbers = to_addresses.get('mobile') or dict()
        # devices = to_addresses.get('device') or list()

        if len(email_addresses) > 1:
            raise BadRequestException(ErrorMessages.ONLY_ONE_TO_EMAIL_SUPPORTED.value)
        if len(mobile_numbers) > 1:
            raise BadRequestException(ErrorMessages.ONLY_ONE_TO_MOBILE_SUPPORTED.value)

        for email in email_addresses:
            if not validate_email(email):
                raise BadRequestException(ErrorMessages.INVALID_EMAIL_FORMAT.value)

        for mobile in mobile_numbers:
            if not validate_mobile(mobile):
                raise BadRequestException(ErrorMessages.INVALID_MOBILE_FORMAT.value)

        if not event.any_channel_active():
            raise BadRequestException(ErrorMessages.NO_CHANNEL_ACTIVE.value)

        # Validate attachments - it must be a list of dicts
        if payload.get("attachments"):
            attachments = payload["attachments"]
            if not isinstance(attachments, list):
                raise BadRequestException(ErrorMessages.INVALID_ATTACHMENT_DATA.value)
            for attachment in attachments:
                if not isinstance(attachment, dict) or 'url' not in attachment or 'filename' not in attachment:
                    raise BadRequestException(ErrorMessages.INVALID_ATTACHMENT_DATA.value)

        # TODO - Need to put validation around verifying the input devices

    @classmethod
    async def trigger_event_notification(cls, payload: dict):
        event_id = str(payload['event_id'])
        event = await Notifications.get_event(event_id)
        if not event:
            raise BadRequestException(ErrorMessages.EVENT_NOT_CONFIGURED.value)

        cls._validate_notification_request(event, payload)

        to_addresses = payload.get('to') or dict()
        email_addresses = to_addresses.get('email') or list()
        mobile_numbers = to_addresses.get('mobile') or dict()
        devices = to_addresses.get('device') or list()

        request_id = generate_uuid()

        final_payload = {
            "request_id": request_id,
            "app_name": event.app_name,
            "event_name": event.event_name,
            "event_id": payload["event_id"],
            "source_identifier": payload["source_identifier"],
            "to": {
                "email": email_addresses,
                "mobile": mobile_numbers,
                "device": devices
            },
            "channels": {
                "email": payload.get("channels", dict()).get("email") or dict(),
                "whatsapp": payload.get("channels", dict()).get("whatsapp") or dict()
            },
            "attachments": payload.get("attachments") or list(),
            "body": payload.get("body") or dict()
        }
        result = await cls.accept_notification_request(event, final_payload)
        return {
            "request_id": request_id, 
            "processing_type": result.processing_type, 
            "message": result.message, 
            **result.extras,
        }

    @classmethod
    async def accept_notification_request(cls, event: Event, data):
        handler = cls.HANDLERS.get(event.priority)
        if not handler:
            raise BadRequestException(ErrorMessages.INVALID_EVENT_PRIORITY.value)

        published = await handler.publish(data)
        if not published.is_success:
            logger.error(
                "Could not publish notification request through handler - %s",
                handler,
            )
            raise BadRequestException(
                "Could not accept this request, please try again after some time"
            )
        return published

app/services/event_notification.py
tata1mg/notifyone-gateway
import asyncio
import copy
import logging

from app.models import Event
from app.service_clients import NotificationServiceClient

logger = logging.getLogger()


class Notifications:

    _lock = asyncio.Lock()
    _sync_delay = 10
    _event_attributes = ['actions', 'event_type', 'meta_info']

    # In-memory cache of app events mapping
    _APP_EVENTS = dict()

    @classmethod
    async def app_events(cls):
        """
        Return app-events mapping from in-memory
        If the mapping is not already initialized, initialize it and return
        As initialization is costly, use asyncio lock to prevent concurrent initializations
        """
        if not cls._APP_EVENTS:
            # Initialize app events mapping
            async with cls._lock:
                if not cls._APP_EVENTS:
                    await cls.initialize_app_events()
                return cls._APP_EVENTS
        else:
            return cls._APP_EVENTS

    @classmethod
    async def refresh_app_events_periodically(cls):
        # sleep for cls._sync_delay seconds
        await asyncio.shield(asyncio.sleep(cls._sync_delay))
        logger.info('Syncing app events after {} seconds'.format(cls._sync_delay))
        # refresh connections list
        try:
            await cls.initialize_app_events()
        except Exception as e:
            logger.error('refresh_app_events_periodically failed with error - {}'.format(str(e)))
        finally:
            # recursive call
            asyncio.create_task(cls.refresh_app_events_periodically())

    @classmethod
    async def initialize_app_events(cls):
        """
        Fetch app events from the downstream, format and load in-memory
        """
        app_events = cls._APP_EVENTS
        all_events = await NotificationServiceClient.get_events_custom(
            attributes=cls._event_attributes
        )
        if all_events:
            app_events = cls._create_app_events_mapping(all_events)
        cls._APP_EVENTS = app_events

    @staticmethod
    def _create_app_events_mapping(events: list) -> dict:
        app_event_mapping = dict()
        for event in events:
            app_event_mapping[str(event['id'])] = copy.deepcopy(event)
        return app_event_mapping

    @classmethod
    async def validate_event(cls, app_name, event_name) -> bool:
        app_events = await cls.app_events()
        if app_name in app_events and event_name in app_events[app_name]:
            return True
        return False

    @classmethod
    async def get_event(cls, event_id:str) -> Event:
        event = None
        app_events = await cls.app_events()
        if event_id in app_events:
            event = Event(app_events[event_id])
        return event

    @classmethod
    async def get_event_notification(cls, notification_request_id):
        return await NotificationServiceClient.get_sent_notifications_by_notification_request_id(notification_request_id)

app/services/notifications.py
tata1mg/notifyone-gateway
from typing import Any, Dict, Optional

import abc
from dataclasses import dataclass

from app.constants import NotificationRequestLogStatus, ProcessingType
from app.utilities.dataclass import JSONSerializableDataClass


@dataclass
class OperatorDetails(JSONSerializableDataClass):
    name: str
    event_id: Optional[str] = None


@dataclass
class PublishResult(JSONSerializableDataClass):
    is_success: bool
    status: NotificationRequestLogStatus
    processing_type: ProcessingType
    message: Any = None
    extras: Optional[Dict[str, Any]] = None

    def __post_init__(self):
        super().__post_init__()
        if self.extras is None:
            self.extras = {}


class Publisher(abc.ABC):
    PROCESSING_TYPE = ProcessingType.UNKNOWN

    @abc.abstractmethod
    async def publish(self, data: Dict) -> PublishResult:
        raise NotImplementedError

app/services/publisher.py
tata1mg/notifyone-gateway
__all__ = ["generate_uuid", "json_dumps", "json_loads"]

from .utils import generate_uuid, json_dumps, json_loads

app/utilities/__init__.py
tata1mg/notifyone-gateway
from typing import ClassVar, List, Optional

from app.utilities.utils import json_dumps
from dataclasses import dataclass, asdict, field


@dataclass
class JSONSerializableDataClass:
    __SERIALIZABLE_KEYS__: ClassVar[Optional[List[str]]] = field(init=False)

    def __post_init__(self):
        ser_keys = getattr(self, "__SERIALIZABLE_KEYS__", None)
        if ser_keys is not None and not isinstance(ser_keys, list):
            raise ValueError("SERIALIZABLE_KEYS must be a list of strings or None")

    @property
    def __dict__(self):
        ser_keys = getattr(self, "__SERIALIZABLE_KEYS__", None)
        data_dict = asdict(self)
        if ser_keys is not None:
            return {key: data_dict[key] for key in self.__SERIALIZABLE_KEYS__}
        return data_dict

    @property
    def json(self):
        return json_dumps(self.__dict__)

app/utilities/dataclass.py
tata1mg/notifyone-gateway
from .rest import RestApiClientWrapper

app/utilities/http/__init__.py
tata1mg/notifyone-gateway
from typing import Any, Dict

from urllib import parse as urlparse

from torpedo.base_http_request import BaseHttpRequest

from app.constants import NotificationRequestLogStatus, ProcessingType
from app.services.publisher import Publisher, PublishResult


class RestApiClientWrapper(Publisher):
    PROCESSING_TYPE = ProcessingType.SYNC

    def __init__(self, host: str, endpoint: str, method: str):
        self._host = host
        self._endpoint = endpoint
        self._method = method

    async def publish(self, payload: Dict[str, Any]) -> PublishResult:
        url = urlparse.urljoin(self._host, self._endpoint)
        response = await BaseHttpRequest.request(
            self._method,
            url,
            data=payload,
            headers={"content-type": "application/json"},
        )
        if response.status == 200:
            publish_result = PublishResult(
                is_success=True,
                processing_type=self.PROCESSING_TYPE,
                status=NotificationRequestLogStatus.SUCCESS,
                message="Message processed via RestClient",
                extras=response.data.get("data"),
            )
        else:
            publish_result = PublishResult(
                is_success=False,
                processing_type=self.PROCESSING_TYPE,
                status=NotificationRequestLogStatus.FAILED,
                message="Something went wrong",
                extras=response.data.get("error", {}).get("message", {}),
            )
        return publish_result

app/utilities/http/rest.py
tata1mg/notifyone-gateway
__all__ = ['SQSWrapper']

from .sqs import SQSWrapper

app/utilities/pubsub/__init__.py
tata1mg/notifyone-gateway
__all__ = ["SQSWrapper"]

from .sqs_wrapper import SQSWrapper

app/utilities/pubsub/sqs/__init__.py
tata1mg/notifyone-gateway
import asyncio
import sys
import zlib
import base64

from app.constants import NotificationRequestLogStatus, ProcessingType
from app.services.publisher import Publisher, PublishResult

from commonutils import BaseSQSWrapper

from app.utilities.utils import json_dumps, json_loads


class SQSWrapper(Publisher):
    PROCESSING_TYPE = ProcessingType.ASYNC

    def __init__(
            self,
            queue_name: str,
            config: dict = {},
            handler: object = None,
            is_compression_enabled: bool = False,
    ):
        config = config or {"SQS": {}}
        self.queue_name = queue_name
        self.sqs_manager = BaseSQSWrapper(config)
        self._create_client_lock = asyncio.Lock()
        self.handler = handler
        self.is_client_created = False
        self.is_compression_enabled = is_compression_enabled

    def __del__(self):
        asyncio.create_task(self.close_client())

    async def close_client(self):
        await self.sqs_manager.close()

    async def init(self):
        if not self.is_client_created:
            async with self._create_client_lock:
                if not self.is_client_created:
                    await self.sqs_manager.get_sqs_client(queue_name=self.queue_name)
            self.is_client_created = True

    async def publish(self, payload: dict, attributes=None, **kwargs) -> PublishResult:
        await self.init()
        attributes = attributes or dict()
        payload_json = json_dumps(payload)
        if self.is_compression_enabled:
            payload_json = self.compress_message(payload_json)
            attributes['compressedMessage'] = {
                "DataType": "String",
                "StringValue": "yes"
            }
        elif self.check_if_compression_needed(payload_json):
            payload_json = self.compress_message(payload_json)
            attributes['compressedMessage'] = {
                "DataType": "String",
                "StringValue": "yes"
            }
        try:
            response = await self.sqs_manager.publish_to_sqs(
                payload=payload_json, attributes=attributes, batch=False, **kwargs
            )
            return PublishResult(
                is_success=True,
                processing_type=self.PROCESSING_TYPE,
                status=NotificationRequestLogStatus.SUCCESS, 
                message="Message successfully published to SQS",
            )
        except Exception as err:
            return PublishResult(
                is_success=False,
                processing_type=self.PROCESSING_TYPE,
                status=NotificationRequestLogStatus.FAILED, 
                message=str(err),
            )

    def compress_payload(self, payload) -> str:
        payload_str = json_dumps(payload)
        return self.compress_message(payload_str)

    @staticmethod
    def compress_message(message: str):
        compressed_msg = zlib.compress(message.encode("utf-8"), 1)
        encoded_msg = base64.b64encode(compressed_msg).decode("utf-8")
        return encoded_msg

    def decompress_to_payload(self, compressed_msg) -> dict:
        decompressed_msg = self.decompress_message(compressed_msg)
        return json_loads(decompressed_msg)

    @staticmethod
    def decompress_message(compressed_msg: str):
        decoded_msg = base64.b64decode(compressed_msg.encode('utf-8'))
        decompressed_msg = zlib.decompress(decoded_msg).decode('utf-8')
        return decompressed_msg

    @staticmethod
    def check_if_compression_needed(payload_json: str) -> bool:
        """
        As AWS SQS supports messages up to 256KB only, compress if the payload size is > 250KB
        Keeping 6KB is for other details in message like messageAttributes etc.
        """
        if sys.getsizeof(payload_json) > 250*1000:
            return True
        return False

app/utilities/pubsub/sqs/sqs_wrapper.py
tata1mg/notifyone-gateway
import json
import uuid


def generate_uuid():
    return str(uuid.uuid4())


def json_dumps(data):
    return json.dumps(data)


def json_loads(data_str):
    return json.loads(data_str)

app/utilities/utils.py
tata1mg/notifyone-gateway
__all__ = ["validate_email", "validate_mobile"]

from .base import validate_email, validate_mobile
app/utilities/validators/__init__.py
tata1mg/notifyone-gateway
import re

mobile_format = re.compile("""^(\+\d{1,3}[- ]?)?\d{10}$""")
email_format = re.compile(
            r"(^[-!#$%&'*+/=?^_`{}|~0-9A-Z]+(\.[-!#$%&'*+/=?^_`{}|~0-9A-Z]+)*"
            r'|^"([\001-\010\013\014\016-\037!#-\[\]-\177]|\\[\001-011\013\014\016-\177])*"'
            r')@(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+[A-Z]{2,10}\.?$', re.IGNORECASE)


def validate_email(email: str) -> bool:
    for email in email.split(','):
        if not email_format.match(email.strip()):
            return False
    return True


def validate_mobile(mobile: str) -> bool:
    if mobile_format.match(mobile):
        return True
    return False

app/utilities/validators/base.py
tata1mg/notifyone-gateway