Squashed 'utils_v2/' content from commit 294b0cd

git-subtree-dir: utils_v2
git-subtree-split: 294b0cd737319e32035c2af18efc83e31c27acf0

sms/savvy_bulk_sms/async_savvy_bulk_sms.py

@@ -0,0 +1,227 @@
+"""
+
+    AUTHOR:
+
+        Khushal P Soonderji
+
+    DATE:
+
+        Wednesday, 4th Dec., 2024
+
+    OBJECTIVE:
+
+        To be able to send SMSs from Savvy Bulk SMS's API.
+
+    REFERENCES:
+
+        N/A
+
+    DOWNLOADS:
+
+        N/A
+
+    WEB-PORTAL:
+
+        N/A
+
+"""
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                   IMPORT                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# To make sibling directories accessible for imports:
+import sys
+sys.path.append(".")
+sys.path.append("..")
+
+# To make API Calls:
+import httpx
+
+# For debugging:
+from icecream import IceCreamDebugger
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                           MACROS / ONE-TIME INIT                                              ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                VARIABLES                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                  CLASSES                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+class AsyncSavvyBulkSMS:
+
+    def __init__(
+            self,
+            api_key: str,
+            partner_id: str,
+            short_code: str,
+            http_client: httpx.AsyncClient = None,
+            debug = True,
+            debug_prefix = "Savvy SMS | ",
+            debug_only_errors = True
+    ):
+
+        """
+        Sets up an instance of an SMS sender that sends messages through Savvy Bulk SMS's API.
+        :param api_key: The API key received from Savvy Bulk SMS.
+        :param partner_id: Your id with Savvy Bulk SMS.
+        :param short_code: Your short code with Savvy Bulk SMS.
+        :param debug: Whether, or not, you would like to show debugging messages (can be changed on the fly).
+        :param debug_prefix: The prefix text to show with the debug string.
+        :param debug_only_errors: Whether you would like to show all debugging messages or just error messages.
+        """
+
+        # Create the debugging tools:
+        self.__printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
+        if not debug: self.__printer.disable()
+        self.__debug_only_errors = debug_only_errors
+
+        # Accept/create an HTTP client to work with:
+        if http_client: self.__http_client = http_client
+        else: self.__http_client = httpx.AsyncClient(
+            limits = httpx.Limits(
+                max_connections = 100,  # ............ Maximum number of connections allowed in the pool.
+                max_keepalive_connections = 50,  # ... Maximum number of connections that can be kept alive.
+            ),
+            timeout = httpx.Timeout(
+                pool = 120.0,  # .... Time to wait for a free connection from the pool.
+                connect = 2.5,  # ... Time to wait for establishing a connection to the server.
+                write = 10.0,  # .... Time to wait for sending data.
+                read = 2.5  # ....... Time to wait for receiving data.
+            )
+        )
+
+        # Capture the input config:
+        self.__api_key = api_key
+        self.__partner_id = partner_id
+        self.__short_code = short_code
+
+    def enable_debug(self):
+        self.__printer.enable()
+
+    def disable_debug(self):
+        self.__printer.disable()
+
+    def debug_only_errors(self):
+        self.__debug_only_errors = True
+
+    def debug_everything(self):
+        self.__debug_only_errors = False
+
+    async def send_sms(
+            self,
+            recipient_number,
+            message
+    ) -> dict | None:
+
+        """
+        Send one SMS from Savvy Bulk SMS's API.
+        :param recipient_number: The mobile no. of the target recipient.
+        :param message: The message to send to the target recipient.
+        :return:
+        """
+
+        # Construct the basic structure of the response of this method:
+        summary = {
+            "success": False,
+            "info": None,
+            "sender": {
+                "partnerId": self.__partner_id,
+                "shortCode": self.__short_code
+            },
+            "recipient": recipient_number,
+            "message": message,
+            "length": len(message),
+            "raw": None,
+        }
+
+        try:
+
+            # Call the API:
+            api_response = await self.__http_client.get(
+                url = r"https://sms.savvybulksms.com/api/services/sendsms/",
+                params = {
+                    "apikey": self.__api_key,
+                    "partnerID": self.__partner_id,
+                    "message": message,
+                    "shortcode": self.__short_code,
+                    "mobile": recipient_number,
+                }
+            )
+
+            # For a successful API call:
+            if api_response.status_code in [200]:
+                api_json = api_response.json()
+                first_desc = api_json.get("responses", [{}])[0].get("response-description", "N/A").lower().strip()
+                summary["success"] = True if first_desc == "success" else False
+                summary["raw"] = {
+                    "http_code": api_response.status_code,
+                    "response": api_json,
+                }
+
+            # For any other code that indicates some form of failure:
+            else: summary["raw"] = {
+                "http_code": api_response.status_code,
+                "response": api_response.content.decode()
+            }
+
+        # If something goes wrong along the way:
+        except Exception as exception:
+            self.__printer(exception)
+
+        # Done here:
+        return summary
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                               MAIN PROGRAM                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+if __name__ == "__main__":
+
+    import asyncio
+
+    async def main():
+
+        sender = AsyncSavvyBulkSMS(
+            api_key = "<your_api_key>",
+            partner_id = "<your_partner_id>",
+            short_code = "<your_short_code>"
+        )
+
+        response = await sender.send_sms(
+            recipient_number = "<target_recipient>",
+            message = "<message_for_recipient>"
+        )
+        print("SMS API RESPONSE:", response)
+
+
+    asyncio.run(main())