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-handler
import re
from typing import Any, Tuple

import aiohttp

from app.constants import aws as ac


async def get_file_details(url: str, timeout: int = 30) -> Tuple[str, Any]:
    """
    Fetch s3 file content using pre-signed URL
    """
    async with aiohttp.ClientSession() as session:
        response = await session.get(url)
        content = await response.content.read()
        headers = response.headers
        return headers, content

app/commons/aws.py
tata1mg/notifyone-handler
from dataclasses import dataclass
from typing import List, Optional, Union


@dataclass
class Recipients:
    to: List[str]
    cc: List[str]
    bcc: List[str]

    def _convert_to_list(self, emails: Union[str, List[str]]) -> List[str]:
        if not emails:
            return []

        if isinstance(emails, list):
            return emails
        return emails.split(",")

    def __init__(self, to: str, cc: str, bcc: str) -> None:
        self.to = self._convert_to_list(to)
        self.cc = self._convert_to_list(cc)
        self.bcc = self._convert_to_list(bcc)


@dataclass
class Message:
    subject: str
    html_body: Optional[str]
    text_body: Optional[str] = None
    files: Optional[List[str]] = None

app/commons/email.py
tata1mg/notifyone-handler
from dataclasses import dataclass
from typing import Dict, Optional, Union


@dataclass
class Response:
    status_code: Union[int, str]
    data: Optional[Dict] = None
    error: Optional[Dict] = None
    event_id: Optional[str] = None
    meta: Optional[Dict] = None

    def __post_init__(self):
        if not self.data and not self.error:
            raise Exception("data and error both can not be null")


def is_success(status_code: int):
    return 200 <= status_code <= 299

app/commons/http.py
tata1mg/notifyone-handler
from .types import AsyncLogger, AsyncLoggerContextCreator, Logger, LogRecord

app/commons/logging/__init__.py
tata1mg/notifyone-handler
import json
from dataclasses import dataclass
from typing import Dict, Optional

from app.commons import http, time
from app.commons.logging import types
from app.pubsub.sqs import APIClientSQS
from app.services.notifier import Notifier


@dataclass
class SQSLogRecord:
    provider: Notifier
    response: http.Response
    status: Optional[str] = None

    def __post_init__(self):
        if not self.status:
            self.status = "SUCCESS" if self.response.status_code == 200 else "FAILED"


class SQSLogger(types.AsyncLogger):
    def __init__(self, sqs_client: APIClientSQS) -> None:
        self.sqs_client = sqs_client

    @staticmethod
    def __construct_message(
        log: types.LogRecord, sqsrecord: SQSLogRecord, extras: Dict = None
    ):
        extras = extras or {}
        return json.dumps(
            {
                "sent_at": time.now(as_str=True),
                "notification_log_id": log.log_id,
                "status": sqsrecord.status,
                "message": sqsrecord.response.data
                if sqsrecord.response.data
                else sqsrecord.response.error,
                "operator": sqsrecord.provider.__provider__,
                "operator_event_id": sqsrecord.response.event_id,
                "metadata": sqsrecord.response.meta,
                **extras,
            }
        )

    async def log(
        self,
        log: types.LogRecord,
        extras: Optional[Dict] = None,
    ):
        sqsrecord = SQSLogRecord(
            extras.pop("provider"), extras.pop("response"), extras.pop("status", None)
        )
        message = self.__construct_message(log, sqsrecord, extras)
        await self.sqs_client.publish_to_sqs(payload=message, batch=False)


class SQSAioLogger(types.AsyncLoggerContextCreator):
    def __init__(self, config: Dict) -> None:
        self.logger: SQSLogger = None
        self.queue_name = config.get("QUEUE_NAME")
        self.sqs_client = APIClientSQS(config)

        self._client_created = False

    async def __aenter__(self) -> SQSLogger:
        if not self._client_created:
            await self.sqs_client.get_sqs_client(self.queue_name)
            self.logger = SQSLogger(self.sqs_client)
            self._client_created = True

        return self.logger

    async def __aexit__(self, *args):
        """
        noop
        """

app/commons/logging/sqs.py
tata1mg/notifyone-handler
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Dict, Optional


@dataclass
class LogRecord:
    log_id: str


class Logger(ABC):
    @abstractmethod
    def log(self, log: LogRecord, extras: Optional[Dict] = None):
        """
        Abstract method to be called by any logger to log the message
        """


class AsyncLogger(Logger, ABC):
    @abstractmethod
    async def log(self, log: LogRecord, extras: Optional[Dict] = None):
        """
        Abstract method to be called by any logger to log the message
        """


class AsyncLoggerContextCreator(ABC):
    async def __aenter__(self) -> AsyncLogger:
        """
        Async context manager method to return instance of AsyncLogger
        """

    async def __aexit__(self, *args):
        """
        Async context manager method to operator while exiting from context manager
        """

app/commons/logging/types.py
tata1mg/notifyone-handler
from dataclasses import dataclass
from typing import Optional


@dataclass
class Notification:
    body: str
    title: Optional[str] = None
    image: Optional[str] = None

app/commons/push.py
tata1mg/notifyone-handler
from datetime import datetime


def now(as_str=False):
    now = datetime.utcnow()
    if as_str:
        return now.strftime("%B %d, %Y %H:%M:%S")
    return now

app/commons/time.py
tata1mg/notifyone-handler
from .constants import HTTPStatusCodes

app/constants/__init__.py
tata1mg/notifyone-handler
PRESIGNED_S3_URL_REGEX = "[a-zA-z0-9:/\-?]+\.[a-z0-9]+\.[a-z]+\.[a-z]+\/[a-zA-Z0-9_\-/]+\.[a-z]+\?[a-zA-Z0-9\-/:?=]"

app/constants/aws.py
tata1mg/notifyone-handler
from enum import Enum


class EmailGateways(Enum):

    SPARK_POST = "SPARK_POST"
    AWS_SES = "AWS_SES"


class SmsGateways(Enum):

    SMS_COUNTRY = "SMS_COUNTRY"
    PLIVO = "PLIVO"
    AWS_SNS = "AWS_SNS"


class PushGateways(Enum):

    FCM = "FCM"
    APNS = "APNS"


class WhatsAppGateways(Enum):

    INTERAKT = "INTERAKT"

app/constants/channel_gateways.py
tata1mg/notifyone-handler
import json


class Config:
    @staticmethod
    def json_file_to_dict(file: str) -> dict:
        config = None
        with open(file) as config_file:
            config = json.load(config_file)
        return config

    @staticmethod
    def get_config():
        return Config.json_file_to_dict("./config.json")

app/constants/config.py
tata1mg/notifyone-handler
from enum import Enum

X_SHARED_CONTEXT = "X-SHARED-CONTEXT"


class ListenerEventTypes(Enum):
    AFTER_SERVER_START = "after_server_start"
    BEFORE_SERVER_START = "before_server_start"
    BEFORE_SERVER_STOP = "before_server_stop"
    AFTER_SERVER_STOP = "after_server_stop"


class SMSSenderConstant(Enum):
    SMS_COUNTRY = "sms_country"
    NETCORE = "netcore"
    PLIVO = "plivo"
    AWS_SNS = "aws_sns"
    OTP_CHANNEL = "otp"
    TRANSACTIONAL_CHANNEL = "transactional"
    SMS = "sms"
    CALL = "call"


class HTTPStatusCodes(Enum):
    SUCCESS = 200
    CREATED = 201
    NO_CONTENT = 204
    BAD_REQUEST = 400
    UNAUTHORIZED = 401
    FORBIDDEN = 403
    NOT_FOUND = 404
    INTERNAL_ERROR = 500
    TIMEOUT_ERROR = 408

app/constants/constants.py
tata1mg/notifyone-handler
from enum import Enum


class ErrorMessages(Enum):

    error_message = "Something went wrong"


class ApiHandler(Enum):

    HTTP_REQUEST = "Error while request for METHOD - {}, URL - {}, HEADERS - {}, AUTH - {}, PARAMS - {}"


class JsonDecode(Enum):
    DECODE = "Error while decoding json :: {}"

app/constants/error_messages.py
tata1mg/notifyone-handler
import asyncio
from typing import List

from commonutils import BaseSQSWrapper
from commonutils.handlers.sqs import SQSHandler

from app.commons.logging.sqs import SQSAioLogger
from app.constants.config import Config
from app.pubsub.sqs.sms_sqs import APIClientSQS
from app.pubsub.sqs_handler.sqs_handler import (EmailSqsHandler,
                                                PushSqsHandler, SMSSqsHandler,
                                                WhatsappSqsHandler)
from app.service_clients.callback_handler import CallbackLogger
from app.services.email.handler import EmailHandler
from app.services.push.handler import PushHandler
from app.services.sms.handler import SmsHandler
from app.services.whatsapp.handler import WhatsappHandler


class Initialize:
    config = Config.get_config()
    sqs_event_mapping = {
        "SMS": {"client": APIClientSQS, "handler": SMSSqsHandler},
        "EMAIL": {"client": APIClientSQS, "handler": EmailSqsHandler},
        "PUSH": {"client": APIClientSQS, "handler": PushSqsHandler},
        "WHATSAPP": {"client": APIClientSQS, "handler": WhatsappSqsHandler},
    }

    @classmethod
    async def initialize_sqs_subscribers(cls):
        """
        Initialize only enabled subscribers here.
        Out of all the available subscribers in the code we will enable only those subscribers that are listed in the
        service config (config.ENABLED_CHANNELS list)
        """
        enabled_channels = cls.config.get("ENABLED_CHANNELS")
        if not enabled_channels:
            raise Exception("Enable at least one channel")
        for channel, callback in cls.sqs_event_mapping.items():
            if channel not in enabled_channels:
                continue
            sqs = cls.config.get("SQS", {})
            auth = cls.config.get("SQS_AUTH", {})
            channel_config = sqs.get("SUBSCRIBE", {}).get(channel, {})
            max_no_of_messages = channel_config.get("MAX_MESSAGE", 5)
            queue_name = channel_config.get("QUEUE_NAME")
            subscribers_count = channel_config.get("SUBSCRIBERS_COUNT", 1)

            sqs_handler = callback.get("handler")()

            subscribers: List[BaseSQSWrapper] = []
            for _ in range(subscribers_count):
                sqs_client = callback.get("client")({"SQS": auth})
                await sqs_client.get_sqs_client(queue_name)
                subscribers.append(sqs_client)

            cls.create_subscribers(subscribers, sqs_handler, max_no_of_messages)

    @classmethod
    def create_subscribers(
        cls, subscribers: List[BaseSQSWrapper], handler: SQSHandler, max_messages: int
    ):
        for subscriber in subscribers:
            asyncio.create_task(
                subscriber.subscribe_all(handler, max_no_of_messages=max_messages)
            )

    @classmethod
    def initialize_handlers(cls, logger):
        SmsHandler.initialize(config=cls.config, log=logger)
        EmailHandler.initialize(config=cls.config, log=logger)
        PushHandler.initialize(config=cls.config, log=logger)
        WhatsappHandler.initialize(config=cls.config, log=logger)

    @classmethod
    def initialize_callback_logger(cls, logger):
        CallbackLogger.init(logger=logger)

    @classmethod
    def initialize_service_startup_dependencies(cls):
        sqs = cls.config.get("SQS", {})
        auth = cls.config.get("SQS_AUTH", {})
        logging = sqs.get("PUBLISH", {}).get("LOGGING", {})
        logger = SQSAioLogger({"SQS": auth, **logging})

        cls.initialize_handlers(logger)
        cls.initialize_callback_logger(logger)

app/initialize.py
tata1mg/notifyone-handler
__all__ = ["listeners"]

from .listeners import listeners

app/listeners/__init__.py
tata1mg/notifyone-handler
from torpedo.constants import ListenerEventTypes

from app.initialize import Initialize


async def initialize_sqs_subscribers(app, loop):
    await Initialize.initialize_sqs_subscribers()


listeners = [
    (initialize_sqs_subscribers, ListenerEventTypes.AFTER_SERVER_START.value),
]

app/listeners/listeners.py
tata1mg/notifyone-handler
from .sms_sqs import APIClientSQS

app/pubsub/sqs/__init__.py
tata1mg/notifyone-handler
class Response:
    def __init__(self, response: dict):
        if not (response and response.get("Messages")):
            self.messages = []
        else:
            self.messages = [Message(message) for message in response["Messages"]]


class Message:
    def __init__(self, message: dict):
        if not message:
            self.body = None
            self.attributes = []
            self.receipt_handle = None
        else:
            self.body = message["Body"]
            message_attributes = message.get("MessageAttributes", {})
            self.attributes = [
                Attribute(attribute_name, message_attributes[attribute_name])
                for attribute_name in message_attributes
            ]
            self.receipt_handle = message["ReceiptHandle"]


class Attribute:
    def __init__(self, attribute_name: str, attribute_data: dict):
        self.name = attribute_name
        self.value = attribute_data.get("StringValue")

app/pubsub/sqs/model.py
tata1mg/notifyone-handler
from commonutils import BaseSQSWrapper


class APIClientSQS(BaseSQSWrapper):
    sqs_handler = None

    def __init__(self, config: dict = {}) -> None:
        self.config = config or {"SQS": {}}
        super().__init__(self.config)

app/pubsub/sqs/sms_sqs.py
tata1mg/notifyone-handler
import json

from commonutils.handlers.sqs import SQSHandler
from sanic.log import logger

from app.commons.logging.types import LogRecord
from app.constants.error_messages import JsonDecode
from app.services.email.handler import EmailHandler
from app.services.push.handler import PushHandler
from app.services.sms.handler import SmsHandler
from app.services.whatsapp.handler import WhatsappHandler


class SMSSqsHandler(SQSHandler):
    @classmethod
    async def handle_event(cls, data):
        try:
            data = json.loads(data)
        except json.JSONDecodeError as err:
            logger.info(JsonDecode.DECODE.value.format(err))

        to = data.pop("to")
        message = data.pop("message")
        log_info = LogRecord(log_id=data.pop("notification_log_id", "-1"))
        return await SmsHandler.notify(to, message, log_info=log_info, **data)


class EmailSqsHandler(SQSHandler):
    @classmethod
    async def handle_event(self, data):
        try:
            data = json.loads(data)
        except json.JSONDecodeError as e:
            logger.info(JsonDecode.DECODE.value.format(e))
            raise e

        log_info = LogRecord(log_id=data.pop("notification_log_id", "-1"))
        to = data.pop("to", "")
        message = data.pop("message", "")
        return await EmailHandler.notify(
            to=to, message=message, log_info=log_info, **data
        )


class WhatsappSqsHandler(SQSHandler):
    @classmethod
    async def handle_event(cls, data):
        try:
            data = json.loads(data)
        except json.JSONDecodeError as err:
            logger.info(JsonDecode.DECODE.value.format(err))
            raise err

        log_info = LogRecord(data.pop("notification_log_id", "-1"))
        to = data.pop("mobile", "")
        return await WhatsappHandler.notify(to, data, log_info=log_info)


class PushSqsHandler(SQSHandler):
    @classmethod
    async def handle_event(cls, data):
        try:
            data = json.loads(data)
        except json.JSONDecodeError as err:
            logger.error("Not a valid JSON, unable to decode: %s", err)
            raise err

        log_info = LogRecord(log_id=data.get("notification_log_id"))
        data = data.pop("push_data", {})
        registration_ids = []
        for device in data.pop("registered_devices", []):
            registration_ids.append(device.get("register_id"))

        return await PushHandler.notify(
            to=registration_ids, message=data.get("body"), log_info=log_info, **data
        )

app/pubsub/sqs_handler/sqs_handler.py
tata1mg/notifyone-handler
from sanic import Blueprint

from .email import email_notify_bp, email_test_bp, emil_callback_bp
from .push import push_notify_bp, push_test_bp
from .sms import sms_callback_bp, sms_notify_bp, sms_test_bp
from .whatsapp import wa_callback_bp, wa_notify_bp, wa_test_bp

blueprint_group = Blueprint.group(
    emil_callback_bp,
    email_notify_bp,
    email_test_bp,
    sms_callback_bp,
    sms_test_bp,
    sms_notify_bp,
    wa_callback_bp,
    wa_notify_bp,
    wa_test_bp,
    push_notify_bp,
    push_test_bp,
)

app/routes/__init__.py
tata1mg/notifyone-handler
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 OperatorDetails:
    name = openapi.String(
        description="Operator/Gateway name used to send the email", example="<Operator>"
    )
    event_id = openapi.String(
        descritpion="Operator event id for this request", example="12121212121212"
    )


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-handler
from .callback import emil_callback_bp
from .notify import email_notify_bp
from .test import email_test_bp

app/routes/email/__init__.py
tata1mg/notifyone-handler
from torpedo.constants import HTTPMethod
from sanic_openapi import openapi

from app.routes.base_api_model import BaseApiModel, File, OperatorDetails


class Sender:
    name = openapi.String(description="Sender name", example="Tata 1mg", required=True)
    address = openapi.String(
        description="Sender email address", example="xyz@1mg.com", required=True
    )
    reply_to = openapi.String(
        description="Reply to email id of sender",
        example="reply-xyz@1mg.com",
        required=True,
    )


class EmailNotifyApiModel(BaseApiModel):

    _uri = "/notify"
    _name = "notify_email"
    _method = HTTPMethod.POST.value
    _summary = "API to send out email notification"
    _description = (
        "This API is used by the notifyone-core to send out the `critical` priority email notifications. "
        "You can use this API to test out the send email functionality."
    )

    class RequestBodyOpenApiModel:
        event_id = openapi.Integer(
            example=111, required=True, description="Event ID for this request"
        )
        event_name = openapi.String(
            description="Event Name for this request",
            example="test_event",
            required=True,
        )
        app_name = openapi.String(
            description="App Name for this request", example="test_app", required=False
        )
        notification_channel = openapi.String(
            description="Notification channel - `email` for email request",
            example="email",
            required=False,
        )
        notification_log_id = openapi.String(
            description="Log ID generated for this request (in notifyone-core service)",
            example="121212",
            required=True,
        )
        to = openapi.String(
            description="Recipient email ID", example="user@mail.com", required=True
        )
        message = openapi.String(
            description="text/html content for email body",
            example="Hi User <br> This is test email",
            required=True,
        )
        subject = openapi.String(
            description="Content for email subject",
            example="Test email subject",
            required=True,
        )
        sender = Sender
        event_type = openapi.String(
            description="Custom event type. This can be used to write priority logic",
            example="transactional",
            required=False,
        )
        files: openapi.Array(
            File,
            description="List of files to be attached with the email",
            required=False,
        )

    class ResponseBodyOpenApiModel:
        status = openapi.Integer(description="Response status", example=200)
        message = openapi.String(
            description="Message", example="Email sent successfully"
        )
        operator = OperatorDetails


class EmailTestApiModel(BaseApiModel):

    _uri = "/test"
    _name = "test_email"
    _method = HTTPMethod.POST.value
    _summary = "API to send out email notification using SQS subscription code"
    _description = (
        "This API can be used to send out email notification using the SQS subscribe code. "
        "Use this to test the SQS subscribe code"
    )

    class RequestBodyOpenApiModel:
        event_id = openapi.Integer(
            example=111, required=True, description="Event ID for this request"
        )
        event_name = openapi.String(
            description="Event Name for this request",
            example="test_event",
            required=True,
        )
        app_name = openapi.String(
            description="App Name for this request", example="test_app", required=False
        )
        notification_channel = openapi.String(
            description="Notification channel - `email` for email request",
            example="email",
            required=False,
        )
        notification_log_id = openapi.String(
            description="Log ID generated for this request (in notifyone-core service)",
            example="121212",
            required=True,
        )
        to = openapi.String(
            description="Recipient email ID", example="user@mail.com", required=True
        )
        message = openapi.String(
            description="text/html content for email body",
            example="Hi User <br> This is test email",
            required=True,
        )
        subject = openapi.String(
            description="Content for email subject",
            example="Test email subject",
            required=True,
        )
        sender = Sender
        event_type = openapi.String(
            description="Custom event type. This can be used to write priority logic",
            example="transactional",
            required=False,
        )
        files: openapi.Array(
            File,
            description="List of files to be attached with the email",
            required=False,
        )

    class ResponseBodyOpenApiModel(EmailNotifyApiModel.ResponseBodyOpenApiModel):
        status = openapi.Integer(description="Response status", example=200)
        message = openapi.String(
            description="Message", example="Email sent successfully"
        )
        operator = OperatorDetails

app/routes/email/api_model.py
tata1mg/notifyone-handler
from typing import List, Optional, Union

from sanic import Blueprint


class EmailBlueprint(Blueprint):
    def __init__(
        self,
        name: str = None,
        url_prefix: Optional[str] = None,
        host: Optional[Union[List[str], str]] = None,
        version: Optional[Union[int, str, float]] = None,
        strict_slashes: Optional[bool] = None,
        version_prefix: str = "/v",
    ):
        url_prefix = url_prefix or ""
        url_prefix = "email/" + url_prefix
        super(EmailBlueprint, self).__init__(
            name=name,
            url_prefix=url_prefix,
            host=host,
            version=version,
            strict_slashes=strict_slashes,
            version_prefix=version_prefix,
        )

app/routes/email/blueprint.py
tata1mg/notifyone-handler
import logging

from sanic import Request
from sanic.response import json

from app.routes.email.blueprint import EmailBlueprint
from app.routes.middlewares.requests import HttpRequestParser
from app.service_clients.callback_handler import CallbackHandler
from app.services.email.handler import EmailHandler

logger = logging.getLogger()

emil_callback_bp = EmailBlueprint("callback_email", url_prefix="callbacks/")


@emil_callback_bp.on_request
async def request_parser(request: Request):
    """Middleware to parse each incoming callback request according to content-type headers"""
    HttpRequestParser.parse_request(request)


@emil_callback_bp.route("/<provider_name:str>", methods=["POST"])
async def get_callback(req: Request, provider_name: str):
    provider = EmailHandler.PROVIDERS.get(provider_name)

    if not provider:
        return json({"error": f"Provider {provider_name} not found"}, status=404)

    if not isinstance(provider, CallbackHandler):
        logger.error("Given provider %s can't handle callback", provider.__provider__)
        return json(
            {"error": f"Provider {provider.__provider__} can't handle callback"},
            status=400,
        )

    await provider.handle_callback(req.ctx.data)
    return json({"data": f"Callback handled for {provider.__provider__}"}, status=200)

app/routes/email/callback.py
tata1mg/notifyone-handler
from torpedo import Request, send_response

from app.commons.logging.types import LogRecord
from app.routes.email.api_model import EmailNotifyApiModel
from app.routes.email.blueprint import EmailBlueprint
from app.services.email.handler import EmailHandler

email_notify_bp = EmailBlueprint("notify_email")

from sanic_openapi import openapi


@email_notify_bp.route(
    EmailNotifyApiModel.uri(),
    methods=[EmailNotifyApiModel.http_method()],
    name=EmailNotifyApiModel.name(),
)
@openapi.definition(
    summary=EmailNotifyApiModel.summary(),
    description=EmailNotifyApiModel.description(),
    body={
        EmailNotifyApiModel.request_content_type(): EmailNotifyApiModel.RequestBodyOpenApiModel
    },
    response={
        EmailNotifyApiModel.response_content_type(): EmailNotifyApiModel.ResponseBodyOpenApiModel
    },
)
async def notify(request: Request):
    data = request.json
    to = data.pop("to")
    message = data.pop("message")
    log_info = LogRecord(log_id=data.pop("notification_log_id", "-1"))

    provider, response = await EmailHandler.notify(
        to, message, log_info=log_info, **data
    )
    response_data = {
        "status": response.status_code,
        "message": response.data or response.error,
        "operator": {
            "name": provider.__provider__,
            "event_id": response.event_id,
        },
        "meta": response.meta,
    }
    return send_response(
        data=response_data,
        status_code=response.status_code,
    )

app/routes/email/notify.py
tata1mg/notifyone-handler
from sanic_openapi import openapi
from torpedo import Request, send_response

from app.pubsub.sqs_handler.sqs_handler import EmailSqsHandler
from app.routes.email.api_model import EmailTestApiModel
from app.routes.email.blueprint import EmailBlueprint
from app.routes.test_sqs_subs_code import test_sqs_subscribe_code

email_test_bp = EmailBlueprint("test_email")


@email_test_bp.route("/test", methods=["POST"], name="test_email")
@openapi.definition(
    summary=EmailTestApiModel.summary(),
    description=EmailTestApiModel.description(),
    body={
        EmailTestApiModel.request_content_type(): EmailTestApiModel.RequestBodyOpenApiModel
    },
    response={
        EmailTestApiModel.response_content_type(): EmailTestApiModel.ResponseBodyOpenApiModel
    },
)
async def test_email(req: Request):
    status_code, response_data = await test_sqs_subscribe_code(req, EmailSqsHandler)
    return send_response(data=response_data, status_code=status_code)

app/routes/email/test.py
tata1mg/notifyone-handler
import json

from sanic import Request


class Parsers:
    @staticmethod
    def parse_form(request: Request):
        return dict(request.form)

    @staticmethod
    def parse_json(request: Request):
        return request.json

    @staticmethod
    def parse_text(request: Request):
        return request.body.decode("utf-8")


class HttpRequestParser:
    CONTENT_TYPES = {
        "application/x-www-form-urlencoded": Parsers.parse_form,
        "application/json": Parsers.parse_json,
        "text/plain": Parsers.parse_text,
    }

    @classmethod
    def parse_request(cls, request: Request):
        request.ctx.data = {
            "headers": dict(request.headers),
            "args": dict(request.args),
        }

        content_type = request.headers.get("content-type", "").lower()
        parser = cls._get_parser(content_type)
        request.ctx.data.update({"body": parser(request)})

    @classmethod
    def _get_parser(cls, content_type: str):
        for type, parser in cls.CONTENT_TYPES.items():
            # if content_type contains given key type
            if type in content_type:
                return parser
        return lambda req: None

app/routes/middlewares/requests.py
tata1mg/notifyone-handler
from .notify import push_notify_bp
from .test import push_test_bp

app/routes/push/__init__.py
tata1mg/notifyone-handler
from torpedo.constants import HTTPMethod
from sanic_openapi import openapi

from app.routes.base_api_model import BaseApiModel, OperatorDetails


class Device:
    register_id: openapi.String(
        description="Device registration ID", example="XXXXXXXXX", required=True
    )


class PushData:
    title: openapi.String(
        description="Push notification title",
        example="Test push notification",
        required=True,
    )
    body: openapi.String(
        description="Push notification body",
        example="Hi, This is test push notification",
        required=True,
    )
    target = openapi.String(description="Push notification target URL", required=False)
    image = openapi.String(description="Push notification image URL", required=False)
    registered_devices = openapi.Array(Device, required=True)


class PushNotifyApiModel(BaseApiModel):

    _uri = "/notify"
    _name = "notify_push"
    _method = HTTPMethod.POST.value
    _summary = "API to send out push notification"
    _description = (
        "This API is used by the notifyone-core to send out the `critical` priority push notifications. "
        "You can use this API to test out the send push functionality."
    )

    class RequestBodyOpenApiModel:
        event_id = openapi.Integer(
            example=111, required=True, description="Event ID for this request"
        )
        event_name = openapi.String(
            description="Event Name for this request",
            example="test_event",
            required=True,
        )
        app_name = openapi.String(
            description="App Name for this request", example="test_app", required=False
        )
        notification_log_id = openapi.String(
            description="Log ID generated for this request (in notifyone-core service)",
            example="121212",
            required=True,
        )
        event_type = openapi.String(
            description="Custom event type. This can be used to write priority logic",
            example="transactional",
            required=False,
        )
        push_data = PushData

    class ResponseBodyOpenApiModel:
        status = openapi.Integer(description="Response status", example=200)
        message = openapi.String(
            description="Message", example="Push sent successfully"
        )
        operator = OperatorDetails


class PushTestApiModel(BaseApiModel):

    _uri = "/test"
    _name = "test_push"
    _method = HTTPMethod.POST.value
    _summary = "API to send out push notification using SQS subscription code"
    _description = (
        "This API can be used to send out push notification using the SQS subscribe code. "
        "Use this to test the SQS subscribe code"
    )

    class RequestBodyOpenApiModel:
        event_id = openapi.Integer(
            example=111, required=True, description="Event ID for this request"
        )
        event_name = openapi.String(
            description="Event Name for this request",
            example="test_event",
            required=True,
        )
        app_name = openapi.String(
            description="App Name for this request", example="test_app", required=False
        )
        notification_log_id = openapi.String(
            description="Log ID generated for this request (in notifyone-core service)",
            example="121212",
            required=True,
        )
        event_type = openapi.String(
            description="Custom event type. This can be used to write priority logic",
            example="transactional",
            required=False,
        )
        push_data = PushData

    class ResponseBodyOpenApiModel:
        status = openapi.Integer(description="Response status", example=200)
        message = openapi.String(
            description="Message", example="Push sent successfully"
        )
        operator = OperatorDetails

app/routes/push/api_model.py
tata1mg/notifyone-handler
from typing import List, Optional, Union

from sanic import Blueprint


class PushBlueprint(Blueprint):
    def __init__(
        self,
        name: str = None,
        url_prefix: Optional[str] = None,
        host: Optional[Union[List[str], str]] = None,
        version: Optional[Union[int, str, float]] = None,
        strict_slashes: Optional[bool] = None,
        version_prefix: str = "/v",
    ):
        url_prefix = url_prefix or ""
        url_prefix = "push/" + url_prefix
        super(PushBlueprint, self).__init__(
            name=name,
            url_prefix=url_prefix,
            host=host,
            version=version,
            strict_slashes=strict_slashes,
            version_prefix=version_prefix,
        )

app/routes/push/blueprint.py
tata1mg/notifyone-handler
from sanic import Request, json
from sanic_openapi import openapi

from app.commons.logging.types import LogRecord
from app.routes.push.api_model import PushNotifyApiModel
from app.routes.push.blueprint import PushBlueprint
from app.services.push.handler import PushHandler

push_notify_bp = PushBlueprint("notify_push")


@push_notify_bp.route(
    PushNotifyApiModel.uri(),
    methods=[PushNotifyApiModel.http_method()],
    name=PushNotifyApiModel.name(),
)
@openapi.definition(
    summary=PushNotifyApiModel.summary(),
    description=PushNotifyApiModel.description(),
    body={
        PushNotifyApiModel.request_content_type(): PushNotifyApiModel.RequestBodyOpenApiModel
    },
    response={
        PushNotifyApiModel.response_content_type(): PushNotifyApiModel.ResponseBodyOpenApiModel
    },
)
async def notify(request: Request):
    data = request.json
    to = data.pop("to")
    message = data.pop("body")
    log_info = LogRecord(log_id=data.get("notification_log_id"))
    registration_ids = []
    for device in data.pop("registered_devices", []):
        registration_ids.append(device.get("register_id"))

    provider, response = await PushHandler.notify(
        to, message, log_info=log_info, **data
    )

    return json(
        {
            "status": response.status_code,
            "message": response.data or response.error,
            "operator": {
                "name": provider.__provider__,
                "event_id": response.event_id,
            },
            "meta": response.meta,
        },
        status=response.status_code,
    )

app/routes/push/notify.py
tata1mg/notifyone-handler
from sanic_openapi import openapi
from torpedo import Request, send_response

from app.pubsub.sqs_handler.sqs_handler import PushSqsHandler
from app.routes.push.api_model import PushTestApiModel
from app.routes.push.blueprint import PushBlueprint
from app.routes.test_sqs_subs_code import test_sqs_subscribe_code

push_test_bp = PushBlueprint("test_push")


@push_test_bp.route(
    PushTestApiModel.uri(),
    methods=[PushTestApiModel.http_method()],
    name=PushTestApiModel.name(),
)
@openapi.definition(
    summary=PushTestApiModel.summary(),
    description=PushTestApiModel.description(),
    body={
        PushTestApiModel.request_content_type(): PushTestApiModel.RequestBodyOpenApiModel
    },
    response={
        PushTestApiModel.response_content_type(): PushTestApiModel.ResponseBodyOpenApiModel
    },
)
async def test_push(req: Request):
    status_code, response_data = await test_sqs_subscribe_code(req, PushSqsHandler)
    return send_response(data=response_data, status_code=status_code)

app/routes/push/test.py
tata1mg/notifyone-handler
from .callback import sms_callback_bp
from .notify import sms_notify_bp
from .test import sms_test_bp

app/routes/sms/__init__.py
tata1mg/notifyone-handler
from torpedo.constants import HTTPMethod
from sanic_openapi import openapi

from app.routes.base_api_model import BaseApiModel, OperatorDetails


class SmsNotifyApiModel(BaseApiModel):

    _uri = "/notify"
    _name = "notify_sms"
    _method = HTTPMethod.POST.value
    _summary = "API to send out sms notification"
    _description = (
        "This API is used by the notifyone-core to send out the `critical` priority sms notifications. "
        "You can use this API to test out the send sms functionality."
    )

    class RequestBodyOpenApiModel:
        event_id = openapi.Integer(
            example=111, required=True, description="Event ID for this request"
        )
        event_name = openapi.String(
            description="Event Name for this request",
            example="test_event",
            required=True,
        )
        app_name = openapi.String(
            description="App Name for this request", example="test_app", required=False
        )
        notification_channel = openapi.String(
            description="Notification channel - `sms` for sms request",
            example="sms",
            required=False,
        )
        notification_log_id = openapi.String(
            description="Log ID generated for this request (in notifyone-core service)",
            example="121212",
            required=True,
        )
        to = openapi.String(
            description="Recipient mobile number", example="7827XXXXXX", required=True
        )
        message = openapi.String(
            description="text content for sms body",
            example="Hello, This is a test sms",
            required=True,
        )
        event_type = openapi.String(
            description="Custom event type. This can be used to write priority logic",
            example="transactional",
            required=False,
        )

    class ResponseBodyOpenApiModel:
        status = openapi.Integer(description="Response status", example=200)
        message = openapi.String(description="Message", example="Sms sent successfully")
        operator = OperatorDetails


class SmsTestApiModel(BaseApiModel):

    _uri = "/test"
    _name = "test_sms"
    _method = HTTPMethod.POST.value
    _summary = "API to send out sms notification using SQS subscription code"
    _description = (
        "This API can be used to send out sms notification using the SQS subscribe code. "
        "Use this to test the SQS subscribe code"
    )

    class RequestBodyOpenApiModel:
        event_id = openapi.Integer(
            example=111, required=True, description="Event ID for this request"
        )
        event_name = openapi.String(
            description="Event Name for this request",
            example="test_event",
            required=True,
        )
        app_name = openapi.String(
            description="App Name for this request", example="test_app", required=False
        )
        notification_channel = openapi.String(
            description="Notification channel - `sms` for sms request",
            example="sms",
            required=False,
        )
        notification_log_id = openapi.String(
            description="Log ID generated for this request (in notifyone-core service)",
            example="121212",
            required=True,
        )
        to = openapi.String(
            description="Recipient mobile number", example="7827XXXXXX", required=True
        )
        message = openapi.String(
            description="text content for sms body",
            example="Hello, This is a test sms",
            required=True,
        )
        event_type = openapi.String(
            description="Custom event type. This can be used to write priority logic",
            example="transactional",
            required=False,
        )

    class ResponseBodyOpenApiModel:
        status = openapi.Integer(description="Response status", example=200)
        message = openapi.String(description="Message", example="Sms sent successfully")
        operator = OperatorDetails

app/routes/sms/api_model.py
tata1mg/notifyone-handler
from typing import List, Optional, Union

from sanic import Blueprint


class SmsBlueprint(Blueprint):
    def __init__(
        self,
        name: str = None,
        url_prefix: Optional[str] = None,
        host: Optional[Union[List[str], str]] = None,
        version: Optional[Union[int, str, float]] = None,
        strict_slashes: Optional[bool] = None,
        version_prefix: str = "/v",
    ):
        url_prefix = url_prefix or ""
        url_prefix = "sms/" + url_prefix
        super(SmsBlueprint, self).__init__(
            name=name,
            url_prefix=url_prefix,
            host=host,
            version=version,
            strict_slashes=strict_slashes,
            version_prefix=version_prefix,
        )

app/routes/sms/blueprint.py
tata1mg/notifyone-handler
import logging

from sanic import Blueprint, Request
from sanic.response import json

from app.routes.middlewares.requests import HttpRequestParser
from app.routes.sms.blueprint import SmsBlueprint
from app.service_clients.callback_handler import CallbackHandler
from app.services.sms.handler import SmsHandler

logger = logging.getLogger()

sms_callback_bp = SmsBlueprint("callback_sms", url_prefix="callbacks/")


@sms_callback_bp.on_request
async def request_parser(request: Request):
    """Middleware to parse each incoming callback request according to content-type headers"""
    HttpRequestParser.parse_request(request)


@sms_callback_bp.route("/<provider_name:str>", methods=["POST"])
async def get_callback(req: Request, provider_name: str):
    provider = SmsHandler.PROVIDERS.get(provider_name)

    if not provider:
        return json({"error": f"Provider {provider_name} not found"}, status=404)

    if not isinstance(provider, CallbackHandler):
        logger.error("Given provider %s can't handle callback", provider.__provider__)
        return json(
            {"error": f"Provider {provider.__provider__} can't handle callback"},
            status=400,
        )

    await provider.handle_callback(req.ctx.data)
    return json({"data": f"Callback handled for {provider.__provider__}"}, status=200)

app/routes/sms/callback.py
tata1mg/notifyone-handler
from sanic_openapi import openapi
from torpedo import Request, send_response

from app.commons.logging.types import LogRecord
from app.routes.sms.api_model import SmsNotifyApiModel
from app.routes.sms.blueprint import SmsBlueprint
from app.services.sms.handler import SmsHandler

sms_notify_bp = SmsBlueprint("notify_sms")


@sms_notify_bp.route(
    SmsNotifyApiModel.uri(),
    methods=[SmsNotifyApiModel.http_method()],
    name=SmsNotifyApiModel.name(),
)
@openapi.definition(
    summary=SmsNotifyApiModel.summary(),
    description=SmsNotifyApiModel.description(),
    body={
        SmsNotifyApiModel.request_content_type(): SmsNotifyApiModel.RequestBodyOpenApiModel
    },
    response={
        SmsNotifyApiModel.response_content_type(): SmsNotifyApiModel.ResponseBodyOpenApiModel
    },
)
async def notify(request: Request):
    data = request.json
    to = data.pop("to")
    message = data.pop("message")
    log_info = LogRecord(log_id=data.pop("notification_log_id", "-1"))
    provider, response = await SmsHandler.notify(to, message, log_info=log_info, **data)
    response_data = {
        "status": response.status_code,
        "message": response.data or response.error,
        "operator": {
            "name": provider.__provider__,
            "event_id": response.event_id,
        },
        "meta": response.meta,
    }
    return send_response(
        data=response_data,
        status_code=response.status_code,
    )

app/routes/sms/notify.py
tata1mg/notifyone-handler
from sanic_openapi import openapi
from torpedo import Request, send_response

from app.pubsub.sqs_handler.sqs_handler import SMSSqsHandler
from app.routes.sms.api_model import SmsTestApiModel
from app.routes.sms.blueprint import SmsBlueprint
from app.routes.test_sqs_subs_code import test_sqs_subscribe_code

sms_test_bp = SmsBlueprint("test_sms")


@sms_test_bp.route(
    SmsTestApiModel.uri(),
    methods=[SmsTestApiModel.http_method()],
    name=SmsTestApiModel.name(),
)
@openapi.definition(
    summary=SmsTestApiModel.summary(),
    description=SmsTestApiModel.description(),
    body={
        SmsTestApiModel.request_content_type(): SmsTestApiModel.RequestBodyOpenApiModel
    },
    response={
        SmsTestApiModel.response_content_type(): SmsTestApiModel.ResponseBodyOpenApiModel
    },
)
async def test_sms(req: Request):
    status_code, response_data = await test_sqs_subscribe_code(req, SMSSqsHandler)
    return send_response(data=response_data, status_code=status_code)

app/routes/sms/test.py
tata1mg/notifyone-handler
from commonutils.handlers.sqs import SQSHandler
from torpedo import Request

from app.utils import json_dumps


async def test_sqs_subscribe_code(request: Request, handler: [SQSHandler]):
    payload = request.json or {}
    payload_json = json_dumps(payload)
    provider, response = await handler.handle_event(payload_json)
    response_data = {
        "status": response.status_code,
        "message": response.data or response.error,
        "operator": {
            "name": provider.__provider__,
            "event_id": response.event_id,
        },
        "meta": response.meta,
    }
    return response.status_code, response_data

app/routes/test_sqs_subs_code.py
tata1mg/notifyone-handler
from .callback import wa_callback_bp
from .notify import wa_notify_bp
from .test import wa_test_bp

app/routes/whatsapp/__init__.py
tata1mg/notifyone-handler
from torpedo.constants import HTTPMethod
from sanic_openapi import openapi

from app.routes.base_api_model import BaseApiModel, File, OperatorDetails


class WhatsappNotifyApiModel(BaseApiModel):

    _uri = "/notify"
    _name = "notify_whatsapp"
    _method = HTTPMethod.POST.value
    _summary = "API to send out whatsapp notification"
    _description = (
        "This API is used by the notifyone-core to send out the `critical` priority whatsapp notifications. "
        "You can use this API to test out the send whatsapp functionality."
    )

    class RequestBodyOpenApiModel:
        event_id = openapi.Integer(
            example=111, required=True, description="Event ID for this request"
        )
        event_name = openapi.String(
            description="Event Name for this request",
            example="test_event",
            required=True,
        )
        app_name = openapi.String(
            description="App Name for this request", example="test_app", required=False
        )
        event_type = openapi.String(
            description="Custom event type. This can be used to write priority logic",
            example="transactional",
            required=False,
        )
        notification_log_id = openapi.String(
            description="Log ID generated for this request (in notifyone-core service)",
            example="121212",
            required=True,
        )
        template = openapi.String(
            description="Whatsapp message template name in Interkt",
            example="order_details",
            required=True,
        )
        mobile = openapi.String(
            description="Recipient mobile number", example="7827XXXXXX", required=True
        )
        body_values = dict(
            description="Values for dynamic place holders in the template",
            example={"order_id": "PO112121"},
            required=False,
        )
        files: openapi.Array(
            File,
            description="List of files to be attached with the message. Currently, only 1 file attachment is supported",
            required=False,
        )

    class ResponseBodyOpenApiModel:
        status = openapi.Integer(description="Response status", example=200)
        message = openapi.String(
            description="Message", example="Whatsapp sent successfully"
        )
        operator = OperatorDetails


class WhatsappTestApiModel(BaseApiModel):

    _uri = "/test"
    _name = "test_whatsapp"
    _method = HTTPMethod.POST.value
    _summary = "API to send out whatsapp notification using SQS subscription code"
    _description = (
        "This API can be used to send out whatsapp notification using the SQS subscribe code. "
        "Use this to test the SQS subscribe code"
    )

    class RequestBodyOpenApiModel:
        event_id = openapi.Integer(
            example=111, required=True, description="Event ID for this request"
        )
        event_name = openapi.String(
            description="Event Name for this request",
            example="test_event",
            required=True,
        )
        app_name = openapi.String(
            description="App Name for this request", example="test_app", required=False
        )
        event_type = openapi.String(
            description="Custom event type. This can be used to write priority logic",
            example="transactional",
            required=False,
        )
        notification_log_id = openapi.String(
            description="Log ID generated for this request (in notifyone-core service)",
            example="121212",
            required=True,
        )
        template = openapi.String(
            description="Whatsapp message template name in Interkt",
            example="order_details",
            required=True,
        )
        mobile = openapi.String(
            description="Recipient mobile number", example="7827XXXXXX", required=True
        )
        body_values = dict(
            description="Values for dynamic place holders in the template",
            example={"order_id": "PO112121"},
            required=False,
        )
        files: openapi.Array(
            File,
            description="List of files to be attached with the message. Currently, only 1 file attachment is supported",
            required=False,
        )

    class ResponseBodyOpenApiModel:
        status = openapi.Integer(description="Response status", example=200)
        message = openapi.String(
            description="Message", example="Whatsapp sent successfully"
        )
        operator = OperatorDetails

app/routes/whatsapp/api_model.py
tata1mg/notifyone-handler
from typing import List, Optional, Union

from sanic import Blueprint


class WhatsappBlueprint(Blueprint):
    def __init__(
        self,
        name: str = None,
        url_prefix: Optional[str] = None,
        host: Optional[Union[List[str], str]] = None,
        version: Optional[Union[int, str, float]] = None,
        strict_slashes: Optional[bool] = None,
        version_prefix: str = "/v",
    ):
        url_prefix = url_prefix or ""
        url_prefix = "whatsapp/" + url_prefix
        super(WhatsappBlueprint, self).__init__(
            name=name,
            url_prefix=url_prefix,
            host=host,
            version=version,
            strict_slashes=strict_slashes,
            version_prefix=version_prefix,
        )

app/routes/whatsapp/blueprint.py
tata1mg/notifyone-handler
import logging

from sanic import Blueprint, Request
from sanic.response import json

from app.routes.middlewares.requests import HttpRequestParser
from app.routes.whatsapp.blueprint import WhatsappBlueprint
from app.service_clients.callback_handler import CallbackHandler
from app.services.whatsapp.handler import WhatsappHandler

logger = logging.getLogger()

wa_callback_bp = WhatsappBlueprint("callback_wa", url_prefix="callbacks/")


@wa_callback_bp.on_request
async def request_parser(request: Request):
    """Middleware to parse each incoming callback request according to content-type headers"""
    HttpRequestParser.parse_request(request)


@wa_callback_bp.route("/<provider_name:str>", methods=["POST"])
async def get_callback(req: Request, provider_name: str):
    provider = WhatsappHandler.PROVIDERS.get(provider_name)

    if not provider:
        return json({"error": f"Provider {provider_name} not found"}, status=404)

    if not isinstance(provider, CallbackHandler):
        logger.error("Given provider %s can't handle callback", provider.__provider__)
        return json(
            {"error": f"Provider {provider.__provider__} can't handle callback"},
            status=400,
        )

    await provider.handle_callback(req.ctx.data)
    return json({"data": f"Callback handled for {provider.__provider__}"}, status=200)

app/routes/whatsapp/callback.py
tata1mg/notifyone-handler
from sanic import Request, json
from sanic_openapi import openapi

from app.commons.logging.types import LogRecord
from app.routes.whatsapp.api_model import WhatsappNotifyApiModel
from app.routes.whatsapp.blueprint import WhatsappBlueprint
from app.services.whatsapp.handler import WhatsappHandler

wa_notify_bp = WhatsappBlueprint("notify_wa")


@wa_notify_bp.route(
    WhatsappNotifyApiModel.uri(),
    methods=[WhatsappNotifyApiModel.http_method()],
    name=WhatsappNotifyApiModel.name(),
)
@openapi.definition(
    summary=WhatsappNotifyApiModel.summary(),
    description=WhatsappNotifyApiModel.description(),
    body={
        WhatsappNotifyApiModel.request_content_type(): WhatsappNotifyApiModel.RequestBodyOpenApiModel
    },
    response={
        WhatsappNotifyApiModel.response_content_type(): WhatsappNotifyApiModel.ResponseBodyOpenApiModel
    },
)
async def notify(request: Request):
    data = request.json
    to = data.pop("to")
    message = data.pop("message")
    log_info = LogRecord(log_id=data.pop("notification_log_id", "-1"))

    provider, response = await WhatsappHandler.notify(
        to, message, log_info=log_info, **data
    )

    return json(
        {
            "status": response.status_code,
            "message": response.data or response.error,
            "operator": {
                "name": provider.__provider__,
                "event_id": response.event_id,
            },
            "meta": response.meta,
        },
        status=response.status_code,
    )

app/routes/whatsapp/notify.py
tata1mg/notifyone-handler
from sanic_openapi import openapi
from torpedo import Request, send_response

from app.pubsub.sqs_handler.sqs_handler import WhatsappSqsHandler
from app.routes.test_sqs_subs_code import test_sqs_subscribe_code
from app.routes.whatsapp.api_model import WhatsappTestApiModel
from app.routes.whatsapp.blueprint import WhatsappBlueprint

wa_test_bp = WhatsappBlueprint("test_wa")


@wa_test_bp.route(
    WhatsappTestApiModel.uri(),
    methods=[WhatsappTestApiModel.http_method()],
    name=WhatsappTestApiModel.name(),
)
@openapi.definition(
    summary=WhatsappTestApiModel.summary(),
    description=WhatsappTestApiModel.description(),
    body={
        WhatsappTestApiModel.request_content_type(): WhatsappTestApiModel.RequestBodyOpenApiModel
    },
    response={
        WhatsappTestApiModel.response_content_type(): WhatsappTestApiModel.ResponseBodyOpenApiModel
    },
)
async def test_wa(req: Request):
    status_code, response_data = await test_sqs_subscribe_code(req, WhatsappSqsHandler)
    return send_response(data=response_data, status_code=status_code)

app/routes/whatsapp/test.py
tata1mg/notifyone-handler
from torpedo import Host

from app.initialize import Initialize
from app.listeners import listeners
from app.routes import blueprint_group

if __name__ == "__main__":
    # register listeners
    Host._listeners = listeners

    # 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.
    Initialize.initialize_service_startup_dependencies()
    Host._blueprint_group = blueprint_group
    Host.run()

app/service.py
tata1mg/notifyone-handler
from typing import Optional

from aiohttp import ClientSession, TCPConnector
from torpedo.exceptions import HTTPRequestException

from app.constants import HTTPStatusCodes
from app.utils import json_dumps


class APIClient:
    SESSION = None
    timeout: int = 30
    headers: dict = None
    auth: tuple = None
    params: dict = None

    @classmethod
    async def get_session(cls):
        if cls.SESSION is None:
            conn = TCPConnector(limit=(0), limit_per_host=(0))
            cls.SESSION = ClientSession(connector=conn)
        return cls.SESSION

    async def request(
        self,
        method: str,
        path: str,
        data: Optional[dict] = None,
        json: Optional[dict] = None,
    ):
        response = None
        request_info = {
            "url": path,
            "data": json_dumps(data),
            "json": json,
            "method": method,
            "timeout": self.timeout,
        }
        if self.headers:
            request_info["headers"] = self.headers
        if self.auth:
            request_info["auth"] = self.auth
        if self.params:
            request_info["params"] = self.params
        try:
            session = await self.get_session()
            response = await session.request(**request_info)
        except Exception as err:
            raise HTTPRequestException(
                err,
                HTTPStatusCodes.INTERNAL_ERROR.value,
                "Encountered error while making request",
            ) from err

        return response

app/service_clients/api_handler.py
tata1mg/notifyone-handler
import aiobotocore


class AWSSNSManager:

    _sns_client = None

    @classmethod
    def initialize(cls, config):
        region_name = config["AWS_SNS"]["REGION_NAME"]
        service_name = config["AWS_SNS"]["SERVICE_NAME"]
        session = aiobotocore.session.get_session()
        cls._sns_client = session.create_client(
            service_name=service_name, region_name=region_name
        )

    @classmethod
    async def send_sms(cls, phone_number, message, message_attributes):
        return await cls._sns_client.publish(
            PhoneNumber=phone_number,
            Message=message,
            MessageAttributes=message_attributes,
        )

app/service_clients/aws_sns_manager.py
tata1mg/notifyone-handler
from abc import ABC, abstractmethod
from typing import Any, Dict

from app.commons.logging import AsyncLoggerContextCreator


class CallbackHandler(ABC):
    @classmethod
    @abstractmethod
    async def handle_callback(cls, data: Dict[str, Any]):
        """
        Abstract method to handle callbacks
        """


class CallbackLogger:

    logger = None

    @classmethod
    def init(cls, logger: AsyncLoggerContextCreator):
        cls.logger = logger

app/service_clients/callback_handler.py
tata1mg/notifyone-handler
from typing import Dict

import httpx


class HTTP20Client:
    def __init__(self, cert=None, verify=None) -> None:
        self.client = httpx.AsyncClient(http2=True, cert=cert, verify=verify)

    async def request(
        self,
        method: str,
        url: str,
        files: Dict = None,
        data: Dict = None,
        json: Dict = None,
    ):
        if data is not None and json is not None:
            raise ValueError("data and json both can not be not null")

        async with self.client as client:
            return await client.request(
                method=method, url=url, json=json, data=data, files=files
            )

app/service_clients/http_20_client.py
tata1mg/notifyone-handler
import logging
from abc import ABC, abstractmethod
from typing import Dict

from app.commons.logging.types import AsyncLoggerContextCreator, LogRecord
from app.constants import HTTPStatusCodes
from app.services.gateway_priority import PriorityGatewaySelection
from app.services.notifier import Notifier

logger = logging.getLogger()


class AbstractHandler(PriorityGatewaySelection, ABC):
    """
    Base class for implementing the Handler class for any channel.
    For example, Handler classes for email/sms/push/whatsapp channels will extend this class
    """

    logger = None
    CHANNEL = None
    MAX_ATTEMPTS = None
    DEFAULT_PRIORITY: list = None
    PRIORITY_LOGIC: str = None

    PROVIDERS: Dict[str, Notifier] = {}

    @classmethod
    @abstractmethod
    def handler_config_key(cls):
        """
        This abstract method must be overridden by the actual channel manager.
        """
        pass

    @classmethod
    @abstractmethod
    def gateways_class_mapping(cls):
        """
        This abstract method must be overridden by the actual channel manager.
        """
        pass

    @classmethod
    def get_default_priority(cls) -> list:
        return cls.DEFAULT_PRIORITY

    @classmethod
    def get_priority_logic(cls) -> str:
        return cls.PRIORITY_LOGIC

    @classmethod
    def initialize(cls, config: Dict, log: AsyncLoggerContextCreator):
        handler_config = config.get(cls.handler_config_key())

        cls.DEFAULT_PRIORITY = handler_config["DEFAULT_PRIORITY"]
        cls.PRIORITY_LOGIC = handler_config.get("PRIORITY_LOGIC") or None

        gateways = handler_config["GATEWAYS"]

        cls.ENABLED_GATEWAYS_COUNT = len(gateways.keys())

        for gateway_config in gateways.values():
            id = gateway_config["ID"]
            gateway = gateway_config["GATEWAY"]
            cls.PROVIDERS[id] = cls.gateways_class_mapping()[gateway](
                gateway_config["CONFIGURATION"]
            )
        cls.logger = log

    @classmethod
    async def notify(cls, to, message, **kwargs):
        """
        Trigger actual communication request
        ----------
            to: str
                user to which mails is needed to be sent
            message: str
                template
        """
        # If log_info is not present use log_id as `-1``
        log_info = kwargs.pop("log_info", LogRecord(log_id="-1"))
        provider, response = None, None
        for n_attempts in range(cls.ENABLED_GATEWAYS_COUNT):
            gateway = cls.select_gateway(
                n_attempts, cls._get_priority_logic_data(to, **kwargs)
            )
            provider = cls.PROVIDERS[gateway]
            response = await provider.send_notification(to, message, **kwargs)
            async with cls.logger as log:
                await log.log(
                    log_info,
                    extras={
                        "provider": provider,
                        "response": response,
                        "sent_to": to,
                        "attempt_number": n_attempts,
                        "channel": cls.CHANNEL,
                    },
                )
            if response.status_code == HTTPStatusCodes.SUCCESS.value:
                logger.info(
                    "Successfully sent email using %s", provider.__class__.__name__
                )
                break
            else:
                logger.error(
                    "Couldn't send email using %s provider", provider.__class__.__name__
                )
        return provider, response

    @classmethod
    def _get_priority_logic_data(cls, to, **kwargs):
        return {
            "to": to,
            "app_name": kwargs.get("app_name"),
            "event_id": kwargs.get("event_id"),
            "event_name": kwargs.get("event_name"),
            "event_type": kwargs.get("event_type"),
        }

app/services/abstract_handler.py
tata1mg/notifyone-handler
import logging

from app.constants.channel_gateways import EmailGateways
from app.services.abstract_handler import AbstractHandler
from app.services.email.ses import AwsSesHandler
from app.services.email.sparkpost import SparkPostHandler

logger = logging.getLogger()


class EmailHandler(AbstractHandler):
    CHANNEL = "email"
    HANDLER_CONFIG_KEY = "EMAIL_HANDLER"
    PROVIDER_CLASS_MAPPING = {
        EmailGateways.SPARK_POST.value: SparkPostHandler,
        EmailGateways.AWS_SES.value: AwsSesHandler,
    }

    @classmethod
    def handler_config_key(cls):
        return cls.HANDLER_CONFIG_KEY

    @classmethod
    def gateways_class_mapping(cls):
        return cls.PROVIDER_CLASS_MAPPING

app/services/email/handler.py
tata1mg/notifyone-handler
import asyncio
import logging
from email.mime.application import MIMEApplication
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from typing import Dict, List

import aiobotocore

from app.commons import aws, email, http
from app.constants.channel_gateways import EmailGateways
from app.services.notifier import Notifier

logger = logging.getLogger()


class AwsSesHandler(Notifier):
    __provider__ = EmailGateways.AWS_SES.value

    CHARSET = "UTF-8"
    # ENDPOINT = "http://localhost:4566"  # For local testing

    def __init__(self, config: Dict):
        self.config = config
        self.session = aiobotocore.session.get_session()

    def _get_destination(
        self, recipients: List[str], cc: List[str], bcc: List[str]
    ) -> Dict:
        destination = {"ToAddresses": recipients}

        if cc and isinstance(cc, list):
            destination.update({"CcAddresses": cc})

        if bcc and isinstance(bcc, list):
            destination.update({"BccAddresses": bcc})

        return destination

    def _get_message(self, subject: str, html_body: str, text_body: str = None) -> Dict:
        message = {
            "Body": {
                "Html": {
                    "Charset": self.CHARSET,
                    "Data": html_body,
                }
            },
            "Subject": {
                "Charset": self.CHARSET,
                "Data": subject,
            },
        }
        if text_body:
            message["Body"].update(
                {
                    "Text": {
                        "Charset": self.CHARSET,
                        "Data": text_body,
                    },
                }
            )

        return message

    async def _prepare_multipart_mime(
        self, recipients: email.Recipients, message: email.Message, sender, reply_to
    ):
        multipart_message = MIMEMultipart("mixed")
        multipart_message["Subject"] = message.subject
        multipart_message["From"] = sender
        multipart_message["To"] = ", ".join(recipients.to)

        destinations = recipients.to

        if recipients.cc:
            multipart_message["Cc"] = ", ".join(recipients.cc)
            destinations += recipients.cc
        if recipients.bcc:
            multipart_message["Bcc"] = ", ".join(recipients.bcc)
            destinations += recipients.bcc

        multipart_message.add_header("Reply-to", reply_to)

        message_body = MIMEMultipart("alternative")
        if message.text_body:
            textpart = MIMEText(
                message.text_body.encode(self.CHARSET), "plain", self.CHARSET
            )
            message_body.attach(textpart)
        if message.html_body:
            htmlpart = MIMEText(
                message.html_body.encode(self.CHARSET), "html", self.CHARSET
            )
            message_body.attach(htmlpart)

        for file in message.files:
            url = file["url"]
            filename = file["filename"]
            headers, content = await aws.get_file_details(url)
            part = MIMEApplication(content)
            part.add_header("Content-Disposition", "attachment", filename=filename)
            multipart_message.attach(part)

        multipart_message.attach(message_body)
        return multipart_message, destinations

    async def _send_raw_email_using_ses(
        self, recipients: email.Recipients, message: email.Message, sender, reply_to
    ):
        multipart_message, destinations = await self._prepare_multipart_mime(
            recipients, message, sender, reply_to
        )
        async with self.session.create_client(
            service_name="ses",
            region_name=self.config["AWS_REGION"],
            # endpoint_url=self.ENDPOINT,  # For local testing
        ) as client:
            try:
                response = await client.send_raw_email(
                    Source=sender,
                    Destinations=destinations,
                    RawMessage={"Data": multipart_message.as_string()},
                )
                return http.Response(
                    status_code=200,
                    data={"data": response},
                    event_id=response["MessageId"],
                    meta=response,
                )
            except Exception as err:
                logger.error("Encountered error while calling send_raw_email: %s", err)
                return http.Response(
                    status_code=400, error={"error": str(err)}, meta=str(err)
                )

    async def _send_email_using_ses(
        self,
        recipients: email.Recipients,
        message: email.Message,
        sender: str,
        reply_to: str,
    ):
        destination = self._get_destination(
            recipients.to, recipients.cc, recipients.bcc
        )
        message = self._get_message(
            message.subject, message.html_body, message.text_body
        )
        async with self.session.create_client(
            service_name="ses",
            region_name=self.config["AWS_REGION"],
            # endpoint_url=self.ENDPOINT,  # For local testing
        ) as client:
            try:
                response = await client.send_email(
                    Destination=destination,
                    Message=message,
                    Source=sender,
                    ReplyToAddresses=[reply_to],
                )
                return http.Response(
                    status_code=200,
                    data={"data": response},
                    event_id=response["MessageId"],
                    meta=response,
                )
            except Exception as err:
                logger.error("Encountered error while calling send_email: %s", err)
                return http.Response(
                    status_code=400, error={"error": str(err)}, meta=str(err)
                )

    async def _send_email_without_attachments(
        self,
        to: str,
        body: str,
        subject: str,
        reply_to: str,
        sender: dict = None,
        **kwargs,
    ):
        sender_address = "{name}<{address}>".format(
            name=sender["name"], address=sender["address"]
        )

        cc = kwargs.get("cc")
        bcc = kwargs.get("bcc")
        recipients = email.Recipients(to, cc, bcc)
        message = email.Message(subject, html_body=body)
        try:
            return await self._send_email_using_ses(
                recipients=recipients,
                message=message,
                sender=sender_address,
                reply_to=reply_to,
            )
        except Exception as err:
            logger.exception("Couldn't send mail using AWS SES %s", err)

    async def _send_email_with_attachments(
        self,
        to: str,
        body: str,
        subject: str,
        files: List[str],
        sender: dict = None,
        **kwargs,
    ):
        """
        Send email using AWS Lambda to attach files with email body
        """
        cc = kwargs.get("cc")
        bcc = kwargs.get("bcc")

        sender_address = "{name}<{address}>".format(
            name=sender["name"], address=sender["address"]
        )

        recipients = email.Recipients(to, cc, bcc)
        message = email.Message(subject, html_body=body, files=files)
        try:
            return await self._send_raw_email_using_ses(
                recipients, message, sender=sender_address, reply_to=sender["reply_to"]
            )
        except Exception as err:
            logger.error("Couldn't send email with attachment using AWS SES: %s", err)
            return http.Response(
                status_code=400, error={"error": str(err)}, meta=str(err)
            )

    async def send_notification(self, to: str, message: str, **kwargs) -> http.Response:
        send_email = self._send_email_without_attachments
        if kwargs.get("files"):  # If files are to be attached
            send_email = self._send_email_with_attachments

        return await send_email(to, message, **kwargs)

app/services/email/ses.py
tata1mg/notifyone-handler
import base64
import logging
from typing import Any, Dict, List
from urllib import parse

from aiohttp import ClientResponse

from app.commons import aws, http
from app.commons.logging.types import LogRecord
from app.constants import HTTPStatusCodes
from app.constants.channel_gateways import EmailGateways
from app.service_clients.api_handler import APIClient
from app.service_clients.callback_handler import (CallbackHandler,
                                                  CallbackLogger)
from app.services.notifier import Notifier
from app.utils import get_value_by_priority

logger = logging.getLogger()


class SparkPostHandler(Notifier, APIClient, CallbackHandler):
    __provider__ = EmailGateways.SPARK_POST.value

    BASE_URL: str = "https://api.sparkpost.com"
    ENDPOINT: str = "/api/v1/transmissions?num_rcpt_errors=3"

    def __init__(self, config: Dict) -> None:
        self.config = config
        self._headers = {
            "Content-Type": "application/json",
            "accept": "application/json",
            "Authorization": self.config["API_KEY"],
        }

    @property
    def headers(self):
        return self._headers

    @headers.setter
    def headers(self, value):
        self._headers.update(value)

    def _get_email_body(self, to: str, body: str, subject: str, sender: dict):
        reply_to = sender["reply_to"]
        email_from = {"email": sender["address"], "name": sender["name"]}
        recipients = []
        to = to.split(",") if isinstance(to, str) else to
        for _email_to in to:
            recipients.append({"address": _email_to})
        return {
            "recipients": recipients,
            "content": {
                "from": email_from,
                "subject": subject,
                "reply_to": reply_to,
                "html": body,
            },
        }

    def _get_url(self):
        return parse.urljoin(self.BASE_URL, self.ENDPOINT)

    async def _send_email_without_attachments(
        self, to, body, subject, sender: dict = None, **kwargs
    ):
        """
        Send email using SparkPost client API
        """
        url = self._get_url()
        payload = self._get_email_body(to, body, subject, sender)
        try:
            response = await self.request(method="post", path=url, data=payload)
            return await self.format_response(response)
        except Exception as err:
            logger.exception(
                "Encountered error while sending email %s, to %s, subject %s",
                err,
                to,
                subject,
            )
            return http.Response(
                status_code=HTTPStatusCodes.INTERNAL_ERROR.value,
                error={"error": f"Encountered error while sending email {str(err)}"},
            )

    @staticmethod
    async def format_response(response: ClientResponse):
        json = await response.json()
        if http.is_success(response.status):
            response = {
                "status_code": HTTPStatusCodes.SUCCESS.value,
                "data": {"data": json},
                "event_id": json["results"]["id"],
                "meta": json,
            }
        else:
            response = {
                "status_code": HTTPStatusCodes.BAD_REQUEST.value,
                "error": {"error": json},
                "meta": json,
            }
        return http.Response(**response)

    async def _send_email_with_attachments(
        self,
        to: str,
        body: str,
        subject: str,
        files: List[str],
        sender: dict = None,
        **kwargs,
    ):
        """
        Send email using AWS Lambda to attach files with email body
        """
        attachments = []
        for file in files:
            url = file["url"]
            filename = file["filename"]
            headers, content = await aws.get_file_details(url)
            attachment = {
                "name": filename,
                "type": headers["content-type"],
                "data": base64.b64encode(content).decode("ascii"),
            }
            attachments.append(attachment)

        payload = self._get_email_body(to, body, subject, sender)
        payload["content"]["attachments"] = attachments
        try:
            response = await self.request("POST", path=self._get_url(), data=payload)
            return await self.format_response(response)
        except Exception as err:
            logger.error(
                "Couldn't send email with attachment: %s",
                err,
            )
            return http.Response(
                status_code=HTTPStatusCodes.BAD_REQUEST.value,
                error={"error": f"Encountered error while sending email {str(err)}"},
            )

    async def send_notification(self, to: str, message: str, **kwargs):
        """
        Send email with
        """
        send_email = self._send_email_without_attachments
        if kwargs.get("files"):  # If files are to be attached
            send_email = self._send_email_with_attachments

        return await send_email(to, message, **kwargs)

    @staticmethod
    def __get_callback_status(status: str):
        status_map = {
            "bounce": "FAILED",
            "delivery": "SUCCESS",
            "injection": "SUCCESS",
            "out_of_band": "FAILED",
            "delay": "FAILED",
            "policy_rejection": "FAILED",
            "generation_failure": "FAILED",
            "generation_rejection": "FAILED",
            "open": "SUCCESS",
            "click": "SUCCESS",
        }
        return status_map.get(status, "UNKNOWN")

    @classmethod
    async def handle_callback(cls, data: Dict[str, Any]):
        AVAILABLE_KEYS = [
            "message_event",
            "track_event",
            "ingest_event",
            "gen_event",
            "relay_event",
            "ab_test_event",
            "unsubscribe_event",
        ]
        logrecord = LogRecord(log_id="-1")

        body = data.get("body")
        if not isinstance(body, list):
            body = [body]

        for data in body:
            msys = data.get("msys", {})
            message = get_value_by_priority(msys, AVAILABLE_KEYS, default={})
            type = message.get("type")
            if not isinstance(type, str):
                logger.error(
                    "status type is not of type a string for operator event_id %s",
                    message.get("transmission_id"),
                )
                continue
            status = cls.__get_callback_status(type.lower())
            response_dict = {
                "status_code": HTTPStatusCodes.SUCCESS.value,
                "event_id": message.get("transmission_id"),
                "meta": data,
            }

            if status == "SUCCESS":
                response_dict.update({"data": {"data": {"status": type}}})
            else:
                response_dict.update(
                    {
                        "error": {
                            "error": {
                                "status": type,
                                "error_code": message.get("error_code"),
                                "reason": message.get("raw_reason"),
                            }
                        }
                    }
                )
            response = http.Response(**response_dict)
            async with CallbackLogger.logger as log:
                await log.log(
                    logrecord,
                    extras={
                        "provider": cls,
                        "response": response,
                        "status": status,
                        "channel": "email",
                    },
                )

app/services/email/sparkpost.py
tata1mg/notifyone-handler
from abc import ABC, abstractmethod


class PriorityGatewaySelection(ABC):
    """
    priority_logic
    """

    @classmethod
    @abstractmethod
    def get_priority_logic(cls) -> str:
        """
        This abstract method must be overridden by the channel manager.
        """
        pass

    @classmethod
    @abstractmethod
    def get_default_priority(cls) -> list:
        """
        This abstract method must be overridden by the channel manager.
        """
        pass

    @classmethod
    def select_gateway(cls, n_attempts: int, request_data: dict) -> str:
        """
        Gateway selection for the current send request
        The logic makes use of below attributes to decide the gateway -
            1) priority logic for the channel
            2) default priority for the channel
            3) n_attempts - no of attempt for this request

        * This will be extended to use the gateways current health and other relevant parameters
        to decide the gateway in the future releases.
        * Current Logic -
            - If the priority logic is set, the gateways order is governed by the priority logic output
            - the (n_attempts)th gateway from the priority order is returned for the current request

        """
        priority_logic_expression = cls.get_priority_logic()
        current_priority = cls.get_default_priority()
        if priority_logic_expression:
            # execute the priority logic to get the priority order of gateways
            try:
                # Never delete the 'data' declaration below. It is used in the eval statement.
                data = request_data
                current_priority = eval(priority_logic_expression)
            except Exception:
                # Log the exception and fallback to default priority order
                pass
        total_gateways = min(len(cls.get_default_priority()), len(current_priority))
        return current_priority[n_attempts % total_gateways]

app/services/gateway_priority.py
tata1mg/notifyone-handler
from abc import ABC, abstractmethod

from app.commons import http


class Notifier(ABC):

    __provider__ = "Provider"

    @abstractmethod
    async def send_notification(self, to: str, message: str, **kwargs) -> http.Response:
        """
        This abstract method should be overridden by the provider handler
        """
        raise NotImplementedError("The method is not implemented yet!")

app/services/notifier.py
tata1mg/notifyone-handler
import logging
from typing import Dict, List

from app.commons import http, push
from app.constants import HTTPStatusCodes
from app.constants.channel_gateways import PushGateways
from app.service_clients.http_20_client import HTTP20Client
from app.services.notifier import Notifier

logger = logging.getLogger()


class APNSHandler(Notifier, HTTP20Client):
    __provider__ = PushGateways.APNS.value

    HOST = "https://api.push.apple.com"
    ENDPOINT = "3/device/{token}"

    def __init__(self, config: Dict) -> None:
        self.CERT = config.get("CERT", "")
        self.PRIVATE_KEY = config.get("PRIVATE_KEY", "")
        super(APNSHandler, self).__init__(cert=(self.CERT, self.PRIVATE_KEY))

    @property
    def headers(self):
        return {
            "Content-Type": "application/json",
        }

    def _get_url(self, token):
        return f"{self.HOST}/{self.ENDPOINT.format(token=token)}"

    def _prepare_payload(self, notification: push.Notification, **kwargs):
        return {
            "notification": {
                "body": notification.body,
            },
            **kwargs,
        }

    @staticmethod
    async def format_response(response):
        if http.is_success(response.status):
            response = {
                "status_code": HTTPStatusCodes.SUCCESS.value,
                "data": {"data": await response.json()},
            }
        else:
            response = {
                "status_code": HTTPStatusCodes.BAD_REQUEST.value,
                "error": {"error": await response.json()},
            }
        return http.Response(**response)

    async def push_notification(
        self, to: str, notification: push.Notification, **kwargs
    ):
        payload = self._prepare_payload(notification, **kwargs)
        try:
            response = await self.request("POST", self._get_url(to), json=payload)
            return await self.format_response(response)
        except Exception as err:
            logger.error(
                "Encountered while making request for push notification %s", err
            )
            return http.Response(
                status_code=HTTPStatusCodes.BAD_REQUEST.value,
                error={"error": f"Encountered error while sending email {str(err)}"},
            )

    async def send_notification(self, to: str, message: str, **kwargs):
        title = kwargs.get("title")
        image = kwargs.get("image")
        notification = push.Notification(title=title, body=message, image=image)

        return await self.push_notification(to, notification, **kwargs)

app/services/push/apns.py
tata1mg/notifyone-handler
import json
import logging
from typing import Dict, List

from aiohttp import ContentTypeError

from app.commons import http, push
from app.constants import HTTPStatusCodes
from app.constants.channel_gateways import PushGateways
from app.service_clients.api_handler import APIClient
from app.services.notifier import Notifier

logger = logging.getLogger()


class FCMHandler(Notifier, APIClient):
    __provider__ = PushGateways.FCM.value

    HOST = "https://fcm.googleapis.com"
    ENDPOINT = "fcm/send"

    def __init__(self, config: Dict) -> None:
        self.AUTH_KEY = config.get("AUTH_KEY", "")

    @property
    def headers(self):
        return {
            "Authorization": f"key={self.AUTH_KEY}",
            "Content-Type": "application/json",
        }

    def _get_url(self):
        return f"{self.HOST}/{self.ENDPOINT}"

    def _prepare_payload(
        self, to: List[str], notification: push.Notification, **kwargs
    ):
        return {
            "registration_ids": to,
            "notification": {
                "title": notification.title,
                "body": notification.body,
                "image": notification.image,
            },
            "data": {**kwargs, "provider": "FCM"},
        }

    @staticmethod
    async def format_response(response):
        content = ""
        try:
            content = await response.json()
            event_id = content.get("multicast_id")
        except ContentTypeError:
            content = await response.text()
            logger.error("Received text response :: %s", content)
            event_id = "-1"
        if http.is_success(response.status):
            response = {
                "status_code": HTTPStatusCodes.SUCCESS.value,
                "data": {"data": content},
                "event_id": event_id,
                "meta": content,
            }
        else:
            response = {
                "status_code": HTTPStatusCodes.BAD_REQUEST.value,
                "error": {"error": content},
                "meta": content,
            }
        return http.Response(**response)

    async def push_notification(
        self, to: List[str], notification: push.Notification, **kwargs
    ):
        payload = self._prepare_payload(to, notification, **kwargs)
        try:
            response = await self.request("POST", self._get_url(), data=payload)
            return await self.format_response(response)
        except Exception as err:
            logger.exception(
                "Encountered error while sending push notification %s", err
            )
            return http.Response(
                status_code=HTTPStatusCodes.INTERNAL_ERROR.value,
                error={
                    "error": f"Encountered error while sending push noitifcation {str(err)}"
                },
            )

    async def send_notification(self, to: List[str], message: str, **kwargs):
        title = kwargs.get("title")
        image = kwargs.get("image")
        notification = push.Notification(title=title, body=message, image=image)

        return await self.push_notification(to, notification, **kwargs)

app/services/push/fcm.py
tata1mg/notifyone-handler
import logging

from app.constants.channel_gateways import PushGateways
from app.services.abstract_handler import AbstractHandler
from app.services.push.apns import APNSHandler
from app.services.push.fcm import FCMHandler

logger = logging.getLogger()


class PushHandler(AbstractHandler):
    CHANNEL = "push"
    HANDLER_CONFIG_KEY = "PUSH_HANDLER"
    PROVIDER_CLASS_MAPPING = {PushGateways.FCM.value: FCMHandler}

    @classmethod
    def handler_config_key(cls):
        return cls.HANDLER_CONFIG_KEY

    @classmethod
    def gateways_class_mapping(cls):
        return cls.PROVIDER_CLASS_MAPPING

app/services/push/handler.py
tata1mg/notifyone-handler
import logging

from app.constants.channel_gateways import SmsGateways
from app.services.abstract_handler import AbstractHandler
from app.services.sms.plivo_manager import PlivoHandler
from app.services.sms.sms_country_manager import SMSCountryHandler
from app.services.sms.sns_manager import SnsHandler

logger = logging.getLogger()


class SmsHandler(AbstractHandler):
    CHANNEL = "sms"
    HANDLER_CONFIG_KEY = "SMS_HANDLER"
    PROVIDER_CLASS_MAPPING = {
        SmsGateways.SMS_COUNTRY.value: SMSCountryHandler,
        SmsGateways.PLIVO.value: PlivoHandler,
        SmsGateways.AWS_SNS.value: SnsHandler,
    }

    @classmethod
    def handler_config_key(cls):
        return cls.HANDLER_CONFIG_KEY

    @classmethod
    def gateways_class_mapping(cls):
        return cls.PROVIDER_CLASS_MAPPING

app/services/sms/handler.py
tata1mg/notifyone-handler
from typing import Any, Dict

from aiohttp import BasicAuth

from app.commons.http import Response
from app.commons.logging import LogRecord
from app.constants.channel_gateways import SmsGateways
from app.constants.constants import HTTPStatusCodes
from app.service_clients.api_handler import APIClient
from app.service_clients.callback_handler import (CallbackHandler,
                                                  CallbackLogger)
from app.services.notifier import Notifier


class PlivoHandler(Notifier, APIClient, CallbackHandler):
    __provider__ = SmsGateways.PLIVO.value

    CALL_STATUS = {
        "ringing": 1,
        "in-progress": 2,
        "busy": 3,
        "completed": 4,
        "failed": 5,
        "timeout": 6,
        "no-answer": 7,
    }

    def __init__(self, config):
        self._config = config
        self._headers = {"content-type": "application/json"}
        self._auth = None
        self._timeout = None

    @property
    def headers(self) -> dict:
        return self._headers

    @headers.setter
    def headers(self, value: dict):
        self._headers.update(value)

    @property
    def auth(self) -> BasicAuth:
        return self._auth

    @auth.setter
    def auth(self, value: tuple):
        self._auth = BasicAuth(*value)

    @property
    def timeout(self) -> int:
        return self._timeout

    @timeout.setter
    def timeout(self, value: int):
        self._timeout = value

    @staticmethod
    def extract_job_id(result: dict):
        job_id = result.get("message_uuid", None)
        if not isinstance(job_id, list):
            return None
        return job_id[0]

    async def send_notification(self, to, message, **kwargs):
        url = self._config["PLIVO_SMS_URL"]
        auth_token = self._config["PLIVO_AUTH_TOKEN"]
        auth_id = self._config["PLIVO_AUTH_ID"]
        callback_url = self._config["PLIVO_CALLBACK_URL"]
        self.timeout = self._config.get("TIMEOUT") or 5
        self.auth = (auth_id, auth_token)
        url = url.format(auth_id=auth_id)
        values = {
            "src": self._config["PLIVO_SENDER_ID"],
            "dst": "+91" + to,
            "text": message,
            "url": callback_url,
            "method": "POST",
        }

        try:
            response = await self.request(method="POST", path=url, data=values)
            result = await response.json()
            if str(response.status).startswith("20"):
                response.job_id = self.extract_job_id(result)
                if response.job_id is not None:
                    return Response(
                        status_code=HTTPStatusCodes.SUCCESS.value,
                        data={"data": result},
                        event_id=response.job_id,
                        meta=result,
                    )
            return Response(
                status_code=HTTPStatusCodes.BAD_REQUEST.value,
                error={"error": result},
                meta=result,
            )
        except Exception as err:
            return Response(
                status_code=HTTPStatusCodes.BAD_REQUEST.value,
                error={"error": str(err)},
                meta=str(err),
            )

    @staticmethod
    def __get_callback_status(status: str):
        status_map = {
            "queued": "SUCCESS",
            "sent": "SUCCESS",
            "failed": "FAILED",
            "delivered": "SUCCESS",
            "undelivered": "FAILED",
            "rejected": "FAILED",
        }

        return status_map.get(status, "UNKNOWN")

    @classmethod
    async def handle_callback(cls, data: Dict[str, Any]):
        logrecord = LogRecord(log_id="-1")
        body = data.get("body")
        status = body.get("Status")[0]
        if isinstance(status, str):
            status = cls.__get_callback_status(status.lower())
        response_dict = {
            "status_code": HTTPStatusCodes.SUCCESS.value,
            "event_id": body.get("MessageUUID")[0],
            "meta": data,
        }
        if status == "SUCCESS":
            response_dict.update({"data": {"data": {"status": body.get("Status")[0]}}})
        else:
            response_dict.update(
                {
                    "error": {
                        "error": {
                            "status": body.get("Status")[0],
                            "error_code": body.get("ErrorCode")[0],
                        }
                    }
                }
            )
        response = Response(**response_dict)
        async with CallbackLogger.logger as log:
            await log.log(
                logrecord,
                extras={
                    "provider": cls,
                    "response": response,
                    "status": status,
                    "channel": "sms",
                },
            )

app/services/sms/plivo_manager.py
tata1mg/notifyone-handler
import logging
import re
from typing import Any, Dict

from app.commons.http import Response
from app.commons.logging import LogRecord
from app.constants.channel_gateways import SmsGateways
from app.constants.constants import HTTPStatusCodes, SMSSenderConstant
from app.service_clients.api_handler import APIClient
from app.service_clients.callback_handler import (CallbackHandler,
                                                  CallbackLogger)
from app.services.notifier import Notifier

logger = logging.getLogger()


class SMSCountryHandler(Notifier, APIClient, CallbackHandler):
    __provider__ = SmsGateways.SMS_COUNTRY.value

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.timeout = config.get("TIMEOUT")

    async def send_notification(self, to, message, **kwargs):
        url = self.config.get("SMS_COUNTRY_URL")
        channel = kwargs.get("channel")
        if channel == SMSSenderConstant.OTP_CHANNEL.value:
            values = self.get_otp_channel_values(to, message)
        else:
            values = self.get_transactional_channel_values(to, message)
        try:
            response = await self.request("POST", url, data=values)
            result = await response.text()
            if str(response.status).startswith("20"):
                response.job_id = self.extract_job_id(result)
                if response.job_id is not None:
                    return Response(
                        status_code=HTTPStatusCodes.SUCCESS.value,
                        data={"data": result},
                        event_id=response.job_id,
                        meta=result,
                    )
            return Response(
                status_code=HTTPStatusCodes.BAD_REQUEST.value,
                error={"error": result},
                meta=result,
            )
        except Exception as err:
            logger.error(
                "Encountered error while sending SMS using SMSCountry: %s", err
            )
            return Response(
                status_code=HTTPStatusCodes.BAD_REQUEST.value,
                error={"error": str(err)},
                meta=str(err),
            )

    def get_otp_channel_values(self, to, body):
        return {
            "User": self.config.get("SMS_COUNTRY_OTP_USERNAME"),
            "passwd": self.config.get("SMS_COUNTRY_OTP_PASSWORD"),
            "message": body,
            "mobilenumber": to,
            "mtype": "N",
            "DR": "Y",
            "sid": self.config.get("SMS_COUNTRY_OTP_SENDER_ID"),
        }

    def get_transactional_channel_values(self, to, body):
        return {
            "User": self.config.get("SMS_COUNTRY_USERNAME"),
            "passwd": self.config.get("SMS_COUNTRY_PASSWORD"),
            "message": body,
            "mobilenumber": to,
            "mtype": "N",
            "DR": "Y",
            "sid": self.config.get("SMS_COUNTRY_SENDER_ID"),
        }

    @staticmethod
    def extract_job_id(text: str):
        match = re.findall(r"^OK:(\d+)$", text)
        if len(match) == 1:
            return match[0]
        return None

    @staticmethod
    def __get_callback_status(status: int):
        """
        SMSCountry Callback status
        0 - Message In Queue
        1 - Submitted To Carrier
        2 - Un Delivered
        3 - Delivered
        4 - Expired
        8 - Rejected
        9 - Message Sent
        10 - Opted Out Mobile Number
        11 - Invalid Mobile Number
        """

        status_map = {
            0: "SUCCESS",
            1: "SUCCESS",
            2: "FAILED",
            3: "SUCCESS",
            4: "FAILED",
            8: "FAILED",
            9: "SUCCESS",
            10: "FAILED",
            11: "FAILED",
        }

        return status_map.get(int(status), "UNKNOWN")

    @classmethod
    async def handle_callback(cls, data: Dict[str, Any]):
        logrecord = LogRecord(log_id="-1")
        args = data.get("args")
        status = cls.__get_callback_status(args.get("status")[0])
        response = Response(
            status_code=200,
            data={"data": {"status": args.get("status")[0]}},
            event_id=args.get("jobno")[0],
            meta=data,
        )

        async with CallbackLogger.logger as log:
            await log.log(
                logrecord,
                extras={
                    "provider": cls,
                    "response": response,
                    "status": status,
                    "channel": "sms",
                },
            )

app/services/sms/sms_country_manager.py
tata1mg/notifyone-handler
from app.commons.http import Response
from app.constants.channel_gateways import SmsGateways
from app.constants.constants import HTTPStatusCodes, SMSSenderConstant
from app.service_clients.aws_sns_manager import AWSSNSManager
from app.services.notifier import Notifier


class SnsHandler(Notifier):

    __provider__ = SmsGateways.AWS_SNS.value

    def __init__(self, config):
        self._config = config

    async def send_notification(self, to, message, **kwargs):
        channel = kwargs.get("channel")
        if channel == SMSSenderConstant.OTP_CHANNEL.value:
            message_attributes = self._config["AWS_SNS"]["MESSAGE_ATTRIBUTES"].get(
                "OTP"
            )
        else:
            message_attributes = self._config["AWS_SNS"]["MESSAGE_ATTRIBUTES"].get(
                "TRANSACTIONAL"
            )

        if not to.startswith("+91"):
            to = "+91" + to

        try:
            response = await AWSSNSManager.send_sms(
                phone_number=to, message=message, message_attributes=message_attributes
            )
            if (
                response
                and response.get("ResponseMetadata", {}).get("HTTPStatusCode") == 200
            ):
                if response.get("MessageId"):
                    return Response(
                        status_code=HTTPStatusCodes.SUCCESS.value,
                        data={"data": response.get("MessageId")},
                    )
                else:
                    return Response(
                        status_code=HTTPStatusCodes.BAD_REQUEST.value,
                        error={"error": response},
                    )
        except Exception as e:
            return Response(
                status_code=HTTPStatusCodes.BAD_REQUEST.value, error={"error": e}
            )

app/services/sms/sns_manager.py
tata1mg/notifyone-handler
import logging

from app.constants.channel_gateways import WhatsAppGateways
from app.services.abstract_handler import AbstractHandler
from app.services.whatsapp.interakt import InteraktHandler

logger = logging.getLogger()


class WhatsappHandler(AbstractHandler):
    CHANNEL = "whatsapp"
    HANDLER_CONFIG_KEY = "WHATSAPP_HANDLER"
    PROVIDER_CLASS_MAPPING = {WhatsAppGateways.INTERAKT.value: InteraktHandler}

    @classmethod
    def handler_config_key(cls):
        return cls.HANDLER_CONFIG_KEY

    @classmethod
    def gateways_class_mapping(cls):
        return cls.PROVIDER_CLASS_MAPPING

app/services/whatsapp/handler.py
tata1mg/notifyone-handler
import json
import logging
import os
from typing import Any, Dict

from app.commons import http
from app.commons.logging import LogRecord
from app.constants.channel_gateways import WhatsAppGateways
from app.constants.constants import HTTPStatusCodes
from app.service_clients.api_handler import APIClient
from app.service_clients.callback_handler import (CallbackHandler,
                                                  CallbackLogger)
from app.services.notifier import Notifier

logger = logging.getLogger()


class InteraktHandler(Notifier, APIClient, CallbackHandler):
    __provider__ = WhatsAppGateways.INTERAKT.value

    def __init__(self, config):
        self._config = config
        self._headers = {"Content-Type": "application/json"}

    @property
    def headers(self) -> dict:
        return self._headers

    @headers.setter
    def headers(self, value: dict):
        self._headers.update(value)

    def _get_url(self) -> str:
        """
        Create the URL for interakt from config
        """
        interakt_host = self._config.get("HOST")
        interakt_path = self._config.get("PATH")
        return interakt_host + interakt_path

    def _get_payload(self, phone_number, data):
        """
        This will prepare the payload for sending whatsapp communication using interakt from data given.
        @param data: The dict containing data for preparing payload.
        """
        # NOTE fullPhoneNumber is phone_number with country code
        if not phone_number.startswith("+91"):
            phone_number = "+91" + phone_number

        template = self._get_template(template_data=data)
        payload = {
            "fullPhoneNumber": phone_number,
            "type": "Template",
            "template": template,
        }
        return payload

    def _get_template(self, template_data):
        """
        This is used to get the template required for sending whatsapp messages.
        @template_data: dict containing template, attachment_data (optional),bodyValues etc.
        """
        name_and_language_code_data = self._get_name_and_language_code(
            template=template_data.get("template", "")
        )
        template = {
            "name": name_and_language_code_data.get("name"),
            "languageCode": name_and_language_code_data.get("language_code"),
            "bodyValues": list(template_data.get("body_values")),
        }

        files = template_data.get("files") or list()

        if files:
            # NOTE only one file attachment is sent at time.
            file = files[0]
            attachments = [file["url"]]
            filename = file["filename"]
            template.update({"headerValues": attachments, "fileName": filename})

        return template

    @staticmethod
    def _get_name_and_language_code(template: str):
        """
        This is used to get name , language_code from template given
        Example template = 'template_name:en'
        then here name is template_name and en is language_code
        in case template does not have language_code then default en will be used
        @param template: a str template containing : separated name and language_code
        @return a dict containing name,language_code
        """
        template = template.split(":")

        # default language_code is 'en'
        language_code = "en"

        # CASE language_code provided in template.
        if len(template) > 1:
            language_code = template[1]

        return {"name": template[0], "language_code": language_code}

    async def send_notification(self, to: str, data: Dict, **kwargs) -> http.Response:
        """
        This will send whatsapp messages based on data given.
        @param data : A dict containing template(name:language_code),template body_values.
        @return A Response dict containing status_code,message,id,result etc.
        """
        try:
            self.headers = {"Authorization": self._config.get("AUTHORIZATION")}
            payload = self._get_payload(phone_number=to, data=data)
            send_whatsapp_message_url = self._get_url()
            response = await self.request(
                method="post",
                path=send_whatsapp_message_url,
                data=payload,
            )
            response_status = response.status
            response_text = await response.text()
            response = json.loads(response_text)
            response.update({"status_code": response_status})
            if not str(response_status).startswith("20"):
                return http.Response(
                    status_code=HTTPStatusCodes.BAD_REQUEST.value,
                    error={"error": response},
                    meta=response,
                )
            return http.Response(
                status_code=HTTPStatusCodes.SUCCESS.value,
                data={"data": response},
                event_id=response["id"],
                meta=response,
            )
        except Exception as err:
            logger.exception(
                "Encountered error while sending WhatsApp communication %s", err
            )
            return http.Response(
                status_code=HTTPStatusCodes.BAD_REQUEST.value,
                error={"error": str(err)},
            )

    @staticmethod
    def __get_callback_status(status: str):
        status_mapping = {
            "message_api_sent": "SUCCESS",
            "message_api_delivered": "SUCCESS",
            "message_api_read": "SUCCESS",
            "message_api_failed": "FAILED",
        }

        return status_mapping.get(status, "UNKNOWN")

    @classmethod
    async def handle_callback(cls, data: Dict[str, Any]):
        logrecord = LogRecord(log_id="-1")
        body: Dict[str, Any] = data.get("body")
        status = body.get("type")
        data = body.get("data", {})
        message = data.get("message", {})
        status = cls.__get_callback_status(status)

        response = http.Response(
            status_code=200,
            data={
                "data": {
                    "status": message.get("message_status"),
                    "failure_reason": message.get("channel_failure_reason"),
                }
            },
            event_id=message.get("id"),
            meta=data,
        )

        async with CallbackLogger.logger as log:
            await log.log(
                logrecord,
                extras={
                    "provider": cls,
                    "response": response,
                    "status": status,
                    "channel": "whatsapp",
                },
            )

app/services/whatsapp/interakt.py
tata1mg/notifyone-handler
import json
import logging
from typing import Any, Dict, List, Optional

logger = logging.getLogger()


def log_combined_error(title, exception):
    request_params = {"exception": exception}
    logger.error(title, extra=request_params)
    combined_error = title + " " + exception
    logger.info(combined_error)


def log_combined_exception(title, exception):
    error = "Exception type {} , exception {}".format(type(exception), exception)
    log_combined_error(title, error)


def get_value_by_priority(
    dictionary: Dict[str, Any], keys_to_try: List[str], default: Optional[Any] = None
) -> Any:
    """
    Fetch key's value from a dict from a list of given keys.
    If multiple keys are present, then first one present is returned
    """

    for key in keys_to_try:
        if dictionary.get(key):
            return dictionary[key]
    return default


def json_dumps(json_obj: dict):
    return json.dumps(json_obj)


def json_loads(json_str: str):
    return json.loads(json_str)

app/utils.py
tata1mg/notifyone-handler
import pytest
from torpedo import Host
from torpedo.common_utils import CONFIG, json_file_to_dict

config = CONFIG.config
config_template = json_file_to_dict("./config_template.json")


@pytest.yield_fixture(scope="session")
def app():
    Host._name = config["NAME"]
    _app = Host.get_app()
    _app.update_config(config)
    yield _app


def test_load_from_file(app):
    assert "NAME" in app.config
    assert app.config.NAME == "service-name"
    assert "HOST" in app.config
    assert app.config.HOST == "0.0.0.0"
    assert "POSTGRES_HOST" in app.config
    assert app.config.POSTGRES_HOST not in (None, "")


def test_config_template_sync_with_config_file(app):
    template_keys = set(config_template.keys())
    config_keys = set(config.keys())
    assert (config_keys - template_keys) == set()

tests/test_config.py
tata1mg/notifyone-handler