From 4c47dffc67c8f1ccf69025ca00c8020dde513096 Mon Sep 17 00:00:00 2001 From: Luca Date: Fri, 10 May 2024 18:20:15 +0200 Subject: [PATCH] fix(messaging.backends): do not require inbound settings for outbound and vice versa --- shiftregister/messaging/backends/clicksend.py | 54 +++++++++---------- shiftregister/messaging/backends/sipgate.py | 32 +++++------ 2 files changed, 44 insertions(+), 42 deletions(-) diff --git a/shiftregister/messaging/backends/clicksend.py b/shiftregister/messaging/backends/clicksend.py index 5fdb7a7..5a8419a 100644 --- a/shiftregister/messaging/backends/clicksend.py +++ b/shiftregister/messaging/backends/clicksend.py @@ -2,9 +2,13 @@ import json from ast import literal_eval from datetime import datetime, timezone -from clicksend_client import ApiClient -from clicksend_client import Configuration as BaseConfiguration -from clicksend_client import SMSApi, SmsMessage, SmsMessageCollection +from clicksend_client import ( + ApiClient, + Configuration, + SMSApi, + SmsMessage, + SmsMessageCollection, +) from clicksend_client.rest import ApiException from django.conf import settings from django.core.exceptions import ImproperlyConfigured @@ -19,29 +23,6 @@ __all__ = ("Receiver", "Sender") MAX_BATCH_SIZE = 1000 # see https://developers.clicksend.com/docs/rest/v3/#how-many-messages-can-i-send -class Configuration(BaseConfiguration): - def __init__(self): - super().__init__() - - for setting in ( - "clicksend_password", - "clicksend_sender_id", - "clicksend_username", - ): - try: - settings.SMS_SETTINGS[setting] - except KeyError: - raise ImproperlyConfigured( - f"'{setting}' must be set in SMS_SETTINGS for ClickSend backend" - ) - - self.username = settings.SMS_SETTINGS["clicksend_username"] - self.password = settings.SMS_SETTINGS["clicksend_password"] - - -client = ApiClient(Configuration()) - - class Receiver(BaseReceiver): fetch = None @@ -66,11 +47,30 @@ class Receiver(BaseReceiver): class Sender(BaseSender): + def __init__(self): + for setting in ( + "clicksend_password", + "clicksend_sender_id", + "clicksend_username", + ): + try: + settings.SMS_SETTINGS[setting] + except KeyError: + raise ImproperlyConfigured( + f"'{setting}' must be set in SMS_SETTINGS for ClickSend backend" + ) + + config = Configuration() + config.username = settings.SMS_SETTINGS["clicksend_username"] + config.password = settings.SMS_SETTINGS["clicksend_password"] + + self.client = ApiClient(config) + def send(self, messages): messages = messages[:MAX_BATCH_SIZE] try: - response = SMSApi(client).sms_send_post( + response = SMSApi(self.client).sms_send_post( SmsMessageCollection( messages=[ SmsMessage( diff --git a/shiftregister/messaging/backends/sipgate.py b/shiftregister/messaging/backends/sipgate.py index 57b66fa..9dee189 100644 --- a/shiftregister/messaging/backends/sipgate.py +++ b/shiftregister/messaging/backends/sipgate.py @@ -12,22 +12,18 @@ __all__ = ("Receiver", "Sender") BASE_URL = "https://api.sipgate.com/v2" -inbound_auth = HTTPBasicAuth( - settings.SMS_SETTINGS.get( - "sipgate_incoming_token_id", settings.SIPGATE_INCOMING_TOKEN_ID - ), - settings.SMS_SETTINGS.get( - "sipgate_incoming_token", settings.SIPGATE_INCOMING_TOKEN - ), -) -outbound_auth = HTTPBasicAuth( - settings.SMS_SETTINGS.get("sipgate_token_id", settings.SIPGATE_TOKEN_ID), - settings.SMS_SETTINGS.get("sipgate_token", settings.SIPGATE_TOKEN), -) - class Receiver(BaseReceiver): def __init__(self): + self.auth = HTTPBasicAuth( + settings.SMS_SETTINGS.get( + "sipgate_incoming_token_id", settings.SIPGATE_INCOMING_TOKEN_ID + ), + settings.SMS_SETTINGS.get( + "sipgate_incoming_token", settings.SIPGATE_INCOMING_TOKEN + ), + ) + self.from_dt = None def fetch(self): @@ -50,7 +46,7 @@ class Receiver(BaseReceiver): while offset < total: r = requests.get( f"{BASE_URL}/history", - auth=inbound_auth, + auth=self.auth, params=params | {"offset": offset}, ) r.raise_for_status() @@ -75,11 +71,17 @@ class Receiver(BaseReceiver): class Sender(BaseSender): + def __init__(self): + self.auth = HTTPBasicAuth( + settings.SMS_SETTINGS.get("sipgate_token_id", settings.SIPGATE_TOKEN_ID), + settings.SMS_SETTINGS.get("sipgate_token", settings.SIPGATE_TOKEN), + ) + def send(self, messages): for message in messages: r = requests.post( f"{BASE_URL}/sessions/sms", - auth=outbound_auth, + auth=self.auth, json={ "smsId": settings.SMS_SETTINGS.get( "sipgate_sms_extension", settings.SIPGATE_SMS_EXTENSION