Support for transactional SMS API (#108)

Author: hownowstephen

.github/workflows/main.yml

@@ -4,22 +4,22 @@ on: [ push ]
 
 jobs:
   build:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     strategy:
+      fail-fast: false
       matrix:
-        python: [ '3.6', '3.7', '3.8','3.9' ]
+        python: [ '3.9', '3.10', '3.11', '3.12', '3.13' ]
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v4
     - name: Set up Python
-      uses: actions/setup-python@v2
+      uses: actions/setup-python@v5.6.0
       with:
         python-version: ${{ matrix.python }}
     - name: install dependencies for the minor version
       run: |
         python -m venv venv
         . venv/bin/activate
-        export version=$(grep -v '^$' ${GITHUB_WORKSPACE}/requirements.txt | grep requests | awk -F'=' '{print $2}' )
-        pip install requests==$(echo $version)
+        pip install -r requirements.txt
     - name: run tests
       run: |
         . venv/bin/activate

customerio/__init__.py

@@ -2,5 +2,5 @@
 
 from customerio.client_base import CustomerIOException
 from customerio.track import CustomerIO
-from customerio.api import APIClient, SendEmailRequest, SendPushRequest
+from customerio.api import APIClient, SendEmailRequest, SendPushRequest, SendSMSRequest
 from customerio.regions import Regions

customerio/api.py

@@ -27,6 +27,12 @@ def send_push(self, request):
             request = request._to_dict()
         resp = self.send_request('POST', self.url + "/v1/send/push", request)
         return json.loads(resp)
+    
+    def send_sms(self, request):
+        if isinstance(request, SendSMSRequest):
+            request = request._to_dict()
+        resp = self.send_request('POST', self.url + "/v1/send/sms", request)
+        return json.loads(resp)
 
     # builds the session.
     def _build_session(self):
@@ -211,3 +217,50 @@ def _to_dict(self):
                 data[name] = value
 
         return data
+
+class SendSMSRequest(object):
+    '''An object with all the options avaiable for triggering a transactional push message'''
+    def __init__(self,
+            transactional_message_id=None,
+            to=None,
+            identifiers=None,
+            disable_message_retention=None,
+            send_to_unsubscribed=None,
+            queue_draft=None,
+            message_data=None,
+            send_at=None,
+            language=None,
+        ):
+
+        self.transactional_message_id = transactional_message_id
+        self.to = to
+        self.identifiers = identifiers
+        self.disable_message_retention = disable_message_retention
+        self.send_to_unsubscribed = send_to_unsubscribed
+        self.queue_draft = queue_draft
+        self.message_data = message_data
+        self.send_at = send_at
+        self.language = language
+
+    def _to_dict(self):
+        '''Build a request payload from the object'''
+        field_map = dict(
+            # field name is the same as the payload field name
+            transactional_message_id="transactional_message_id",
+            to="to",
+            identifiers="identifiers",
+            disable_message_retention="disable_message_retention",
+            send_to_unsubscribed="send_to_unsubscribed",
+            queue_draft="queue_draft",
+            message_data="message_data",
+            send_at="send_at",
+            language="language",
+        )
+
+        data = {}
+        for field, name in field_map.items():
+            value = getattr(self, field, None)
+            if value is not None:
+                data[name] = value
+
+        return data

requirements.txt

@@ -1 +1,2 @@
-requests>=2.20.0
+requests>=2.31.0
+urllib3>=2.0.0

tests/server.py

@@ -3,20 +3,18 @@
 except ImportError:
     from http.server import BaseHTTPRequestHandler, HTTPServer
 
-from functools import wraps
 from random import randint
 import json
 import ssl
 import time
 import threading
 import unittest
 
-def sslwrap(func):
-    @wraps(func)
-    def bar(*args, **kw):
-        kw['ssl_version'] = ssl.PROTOCOL_SSLv23
-        return func(*args, **kw)
-    return bar
+def create_ssl_context():
+    """Create SSL context for Python 3.12+ compatibility"""
+    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    context.minimum_version = ssl.TLSVersion.TLSv1_2
+    return context
 
 request_counts = dict()
 
@@ -75,12 +73,11 @@ class HTTPSTestCase(unittest.TestCase):
     def setUpClass(cls):
         # create a server
         cls.server = HTTPServer(("localhost", 0), Handler)
-        # hack needed to setup ssl server
-        ssl.wrap_socket = sslwrap(ssl.wrap_socket)
+        # create SSL context for Python 3.12+ compatibility
+        context = create_ssl_context()
+        context.load_cert_chain('./tests/server.pem')
         # upgrade to https
-        cls.server.socket = ssl.wrap_socket(cls.server.socket,
-            certfile='./tests/server.pem',
-            server_side=True)
+        cls.server.socket = context.wrap_socket(cls.server.socket, server_side=True)
         # start server instance in new thread
         cls.server_thread = threading.Thread(target=cls.server.serve_forever)
         cls.server_thread.start()

tests/test_api.py

@@ -5,7 +5,7 @@
 import sys
 import unittest
 
-from customerio import APIClient, SendEmailRequest, SendPushRequest, Regions, CustomerIOException
+from customerio import APIClient, SendEmailRequest, SendPushRequest, SendSMSRequest, Regions, CustomerIOException
 from customerio.__version__ import __version__ as ClientVersion
 from tests.server import HTTPSTestCase
 
@@ -95,5 +95,21 @@ def test_send_push(self):
 
         self.client.send_push(push)
 
+    def test_send_sms(self):
+        self.client.http.hooks = dict(response=partial(self._check_request, rq={
+            'method': 'POST',
+            'authorization': "Bearer app_api_key",
+            'content_type': 'application/json',
+            'url_suffix': '/v1/send/sms',
+            'body': {"identifiers": {"id":"customer_1"}, "transactional_message_id": 100}
+        }))
+
+        sms = SendSMSRequest(
+            identifiers={"id":"customer_1"},
+            transactional_message_id=100,
+        )
+
+        self.client.send_sms(sms)
+
 if __name__ == '__main__':
     unittest.main()