added configarable sms support

care/emr/api/otp_viewsets/login.py

@@ -10,8 +10,8 @@
 from care.emr.api.viewsets.base import EMRBaseViewSet
 from care.facility.api.serializers.patient_otp import rand_pass
 from care.facility.models import PatientMobileOTP
+from care.utils import sms
 from care.utils.models.validators import mobile_validator
-from care.utils.sms.send_sms import send_sms
 from config.patient_otp_token import PatientToken
 
 
@@ -51,12 +51,12 @@ def send(self, request):
         if settings.USE_SMS:
             random_otp = rand_pass(settings.OTP_LENGTH)
             try:
-                send_sms(
-                    data.phone_number,
-                    (
+                sms.send_text_message(
+                    content=(
                         f"Open Healthcare Network Patient Management System Login, OTP is {random_otp} . "
                         "Please do not share this Confidential Login Token with anyone else"
                     ),
+                    recipients=[data.phone_number],
                 )
             except Exception as e:
                 import logging

care/facility/api/serializers/patient_otp.py

@@ -8,7 +8,7 @@
 from rest_framework.exceptions import ValidationError
 
 from care.facility.models.patient import PatientMobileOTP
-from care.utils.sms.send_sms import send_sms
+from care.utils import sms
 
 
 def rand_pass(size):
@@ -41,12 +41,12 @@ def create(self, validated_data):
         otp = rand_pass(settings.OTP_LENGTH)
 
         if settings.USE_SMS:
-            send_sms(
-                otp_obj.phone_number,
-                (
+            sms.send_text_message(
+                content=(
                     f"Open Healthcare Network Patient Management System Login, OTP is {otp} . "
                     "Please do not share this Confidential Login Token with anyone else"
                 ),
+                recipients=[otp_obj.phone_number],
             )
         elif settings.DEBUG:
             print(otp, otp_obj.phone_number)  # noqa: T201

care/utils/notification_handler.py

@@ -17,7 +17,7 @@
 )
 from care.facility.models.shifting import ShiftingRequest
 from care.users.models import User
-from care.utils.sms.send_sms import send_sms
+from care.utils import sms
 
 logger = logging.getLogger(__name__)
 
@@ -371,10 +371,9 @@ def generate(self):
                 medium == Notification.Medium.SMS.value
                 and settings.SEND_SMS_NOTIFICATION
             ):
-                send_sms(
-                    self.generate_sms_phone_numbers(),
-                    self.generate_sms_message(),
-                    many=True,
+                sms.send_text_message(
+                    content=self.generate_sms_message(),
+                    recipients=self.generate_sms_phone_numbers(),
                 )
             elif medium == Notification.Medium.SYSTEM.value:
                 if not self.message:

care/utils/sms/__init__.py

@@ -0,0 +1,79 @@
+from typing import TYPE_CHECKING
+
+from django.conf import settings
+from django.utils.module_loading import import_string
+
+from care.utils.sms.message import TextMessage
+
+if TYPE_CHECKING:
+    from care.utils.sms.backend.base import SmsBackendBase
+
+
+def initialize_backend(
+    backend_name: str | None = None, fail_silently: bool = False, **kwargs
+) -> "SmsBackendBase":
+    """
+    Load and configure an SMS backend.
+
+    Args:
+        backend_name (Optional[str]): The dotted path to the backend class. If None, the default backend from settings is used.
+        fail_silently (bool): Whether to handle exceptions quietly. Defaults to False.
+
+    Returns:
+        SmsBackendBase: An initialized instance of the specified SMS backend.
+    """
+    backend_class = import_string(backend_name or settings.SMS_BACKEND)
+    return backend_class(fail_silently=fail_silently, **kwargs)
+
+
+def send_text_message(
+    content: str = "",
+    sender: str | None = None,
+    recipients: str | list[str] | None = None,
+    fail_silently: bool = False,
+    backend_instance: type["SmsBackendBase"] | None = None,
+) -> int:
+    """
+    Send a single SMS message to one or more recipients.
+
+    Args:
+        content (str): The message content to be sent. Defaults to an empty string.
+        sender (Optional[str]): The sender's phone number. Defaults to None.
+        recipients (Union[str, List[str], None]): A single recipient or a list of recipients. Defaults to None.
+        fail_silently (bool): Whether to suppress exceptions during sending. Defaults to False.
+        backend_instance (Optional[SmsBackendBase]): A pre-configured SMS backend instance. Defaults to None.
+
+    Returns:
+        int: The number of messages successfully sent.
+    """
+    if isinstance(recipients, str):
+        recipients = [recipients]
+    message = TextMessage(
+        content=content,
+        sender=sender,
+        recipients=recipients,
+        backend=backend_instance,
+        fail_silently=fail_silently,
+    )
+    return message.dispatch(fail_silently=fail_silently)
+
+
+def get_sms_backend(
+    backend_name: str | None = None, fail_silently: bool = False, **kwargs
+) -> "SmsBackendBase":
+    """
+    Load and return an SMS backend instance.
+
+    Args:
+        backend_name (Optional[str]): The dotted path to the backend class. If None, the default backend from settings is used.
+        fail_silently (bool): Whether to suppress exceptions quietly. Defaults to False.
+        **kwargs: Additional arguments passed to the backend constructor.
+
+    Returns:
+        SmsBackendBase: An initialized instance of the specified SMS backend.
+    """
+    return initialize_backend(
+        backend_name=backend_name or settings.SMS_BACKEND,
+        fail_silently=fail_silently,
+        **kwargs,
+    )

care/utils/sms/backend/base.py

@@ -0,0 +1,38 @@
+from care.utils.sms.message import TextMessage
+
+
+class SmsBackendBase:
+    """
+    Base class for all SMS backends.
+
+    Subclasses should override the `send_message` method to provide the logic
+    for sending SMS messages.
+    """
+
+    def __init__(self, fail_silently: bool = False, **kwargs) -> None:
+        """
+        Initialize the SMS backend.
+
+        Args:
+            fail_silently (bool): Whether to suppress exceptions during message sending. Defaults to False.
+            **kwargs: Additional arguments for backend configuration.
+        """
+        self.fail_silently = fail_silently
+
+    def send_message(self, message: TextMessage) -> int:
+        """
+        Send a text message.
+
+        Subclasses must implement this method to handle the logic for sending
+        messages using the specific backend.
+
+        Args:
+            message (TextMessage): The message to be sent.
+
+        Raises:
+            NotImplementedError: If the method is not implemented in a subclass.
+
+        Returns:
+            int: The number of messages successfully sent.
+        """
+        raise NotImplementedError("Subclasses must implement `send_message`.")

care/utils/sms/backend/console.py

@@ -0,0 +1,44 @@
+import sys
+import threading
+
+from care.utils.sms.backend.base import SmsBackendBase
+from care.utils.sms.message import TextMessage
+
+
+class ConsoleBackend(SmsBackendBase):
+    """
+    Outputs SMS messages to the console for debugging purposes.
+    """
+
+    def __init__(self, *args, stream=None, **kwargs) -> None:
+        """
+        Initialize the ConsoleBackend.
+
+        Args:
+            stream (Optional[TextIO]): The output stream to write messages to. Defaults to sys.stdout.
+            *args: Additional arguments for the superclass.
+            **kwargs: Additional keyword arguments for the superclass.
+        """
+        super().__init__(*args, **kwargs)
+        self.stream = stream or sys.stdout
+        self._lock = threading.RLock()
+
+    def send_message(self, message: TextMessage) -> int:
+        """
+        Write the SMS message to the console.
+
+        Args:
+            message (TextMessage): The message to be sent.
+
+        Returns:
+            int: The number of messages successfully "sent" (i.e., written to the console).
+        """
+        sent_count = 0
+        with self._lock:
+            for recipient in message.recipients:
+                self.stream.write(
+                    f"From: {message.sender}\nTo: {recipient}\nContent: {message.content}\n{'-' * 100}\n"
+                )
+                sent_count += 1
+            self.stream.flush()
+        return sent_count

care/utils/sms/backend/sns.py

@@ -0,0 +1,96 @@
+from django.conf import settings
+from django.core.exceptions import ImproperlyConfigured
+
+from care.utils.sms.backend.base import SmsBackendBase
+from care.utils.sms.message import TextMessage
+
+try:
+    import boto3
+    from botocore.exceptions import ClientError
+
+    HAS_BOTO3 = True
+except ImportError:
+    HAS_BOTO3 = False
+
+
+class SnsBackend(SmsBackendBase):
+    """
+    Sends SMS messages using AWS SNS.
+    """
+
+    _sns_client = None
+
+    @classmethod
+    def _get_client(cls):
+        """
+        Get or create the SNS client.
+
+        Returns:
+            boto3.Client: The shared SNS client.
+        """
+        if cls._sns_client is None:
+            region_name = getattr(settings, "SNS_REGION", None)
+
+            if not HAS_BOTO3:
+                raise ImproperlyConfigured(
+                    "Boto3 library is required but not installed."
+                )
+
+            if getattr(settings, "SNS_ROLE_BASED_MODE", False):
+                if not region_name:
+                    raise ImproperlyConfigured(
+                        "AWS SNS is not configured. Check 'SNS_REGION' in settings."
+                    )
+                cls._sns_client = boto3.client(
+                    "sns",
+                    region_name=region_name,
+                )
+            else:
+                access_key_id = getattr(settings, "SNS_ACCESS_KEY", None)
+                secret_access_key = getattr(settings, "SNS_SECRET_KEY", None)
+                if not region_name or not access_key_id or not secret_access_key:
+                    raise ImproperlyConfigured(
+                        "AWS SNS credentials are not fully configured. Check 'SNS_REGION', 'SNS_ACCESS_KEY', and 'SNS_SECRET_KEY' in settings."
+                    )
+                cls._sns_client = boto3.client(
+                    "sns",
+                    region_name=region_name,
+                    aws_access_key_id=access_key_id,
+                    aws_secret_access_key=secret_access_key,
+                )
+        return cls._sns_client
+
+    def __init__(self, fail_silently: bool = False, **kwargs) -> None:
+        """
+        Initialize the SNS backend.
+
+        Args:
+            fail_silently (bool): Whether to suppress exceptions during initialization. Defaults to False.
+            **kwargs: Additional arguments for backend configuration.
+        """
+        super().__init__(fail_silently=fail_silently, **kwargs)
+
+    def send_message(self, message: TextMessage) -> int:
+        """
+        Send a text message using AWS SNS.
+
+        Args:
+            message (TextMessage): The message to be sent.
+
+        Returns:
+            int: The number of messages successfully sent.
+        """
+        sns_client = self._get_client()
+        successful_sends = 0
+
+        for recipient in message.recipients:
+            try:
+                sns_client.publish(
+                    PhoneNumber=recipient,
+                    Message=message.content,
+                )
+                successful_sends += 1
+            except ClientError as error:
+                if not self.fail_silently:
+                    raise error
+        return successful_sends

care/utils/sms/message.py

@@ -0,0 +1,58 @@
+from typing import TYPE_CHECKING
+
+from django.conf import settings
+
+if TYPE_CHECKING:
+    from care.utils.sms.backend.base import SmsBackendBase
+
+
+class TextMessage:
+    """
+    Represents a text message for transmission to one or more recipients.
+    """
+
+    def __init__(
+        self,
+        content: str = "",
+        sender: str | None = None,
+        recipients: list[str] | None = None,
+        backend: type["SmsBackendBase"] | None = None,
+        fail_silently: bool = False,
+    ) -> None:
+        """
+        Initialize a TextMessage instance.
+
+        Args:
+            content (str): The message content.
+            sender (Optional[str]): The sender's phone number.
+            recipients (Optional[List[str]]): List of recipient phone numbers.
+            backend (Optional[SmsBackendBase]): Backend for sending the message.
+        """
+        self.content = content
+        self.sender = sender or getattr(settings, "DEFAULT_SMS_SENDER", "")
+        self.recipients = recipients or []
+        self.backend = backend
+
+        if not self.backend:
+            from care.utils.sms import get_sms_backend
+
+            self.backend = get_sms_backend(fail_silently=fail_silently)
+
+        if isinstance(self.recipients, str):
+            raise ValueError("Recipients should be a list of phone numbers.")
+
+    def dispatch(self, fail_silently: bool = False) -> int:
+        """
+        Send the message to all designated recipients.
+
+        Args:
+            fail_silently (bool): Whether to suppress errors during message sending.
+
+        Returns:
+            int: Count of successfully sent messages.
+        """
+        if not self.recipients:
+            return 0
+
+        connection = self.backend
+        return connection.send_message(self)