Add boilerplate for sending SMS via Reach

This works in conjunction with the new SMS provider stub [^1].

Local testing:

- Run the migrations to add Reach as an inactive provider.
- Activate the Reach provider locally and deactivate the others.

      update provider_details set priority = 100, active = false where notification_type = 'sms';
      update provider_details set active = true where identifier = 'reach';

- Tweak your local environment to point at the SMS stub.

      export REACH_URL="http://host.docker.internal:6300/reach"

- Start / restart Celery to pick up the config change.
- Send a SMS via the Admin app and see the stub log it.
- Reset your environment so you can send normal SMS.

      update provider_details set active = true where notification_type = 'sms';
      update provider_details set active = false where identifier = 'reach';

[^1]: https://github.com/alphagov/notifications-sms-provider-stub/pull/10
This commit is contained in:
Ben Thorner
2022-03-25 15:03:52 +00:00
parent 27ddc4501e
commit 015152bab2
6 changed files with 203 additions and 7 deletions

View File

@@ -1,3 +1,7 @@
import json
from requests import RequestException, request
from app.clients.sms import SmsClient, SmsClientResponseException
@@ -13,13 +17,49 @@ def get_reach_responses(status, detailed_status_code=None):
class ReachClientResponseException(SmsClientResponseException):
pass # TODO (custom exception for errors)
def __init__(self, response, exception):
status_code = response.status_code if response is not None else 504
text = response.text if response is not None else "Gateway Time-out"
self.status_code = status_code
self.text = text
self.exception = exception
def __str__(self):
return "Code {} text {} exception {}".format(self.status_code, self.text, str(self.exception))
class ReachClient(SmsClient):
def init_app(self, *args, **kwargs):
super().init_app(*args, **kwargs)
self.url = self.current_app.config.get('REACH_URL')
def get_name(self):
pass # TODO
@property
def name(self):
return 'reach'
def send_sms(self, to, content, reference, international, multi=True, sender=None):
pass # TODO
def try_send_sms(self, to, content, reference, international, sender):
data = {
# TODO
}
try:
response = request(
"POST",
self.url,
data=json.dumps(data),
headers={
'Content-Type': 'application/json',
},
timeout=60
)
response.raise_for_status()
try:
json.loads(response.text)
except (ValueError, AttributeError) as e:
raise ReachClientResponseException(response=response, exception=e)
except RequestException as e:
raise ReachClientResponseException(response=e.response, exception=e)
return response