darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-08-24 00:06:16 -04:00
Code Issues Packages Projects Releases Wiki Activity

code review feedback

Browse Source
This commit is contained in:
Kenneth Kehl
2024-12-19 07:18:14 -08:00
parent 9e5b21ee30 5a6980f31e
commit 8cd0bd81f7
22 changed files with 226 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
app/aws/s3.py
View File

@@ -402,7 +402,12 @@ def extract_phones(job):
phone_index = 0
for item in first_row:
# Note: may contain a BOM and look like \ufeffphone number
if item.lower() in ["phone number", "\\ufeffphone number"]:
if item.lower() in [
"phone number",
"\\ufeffphone number",
"\\ufeffphone number\n",
"phone number\n",
]:
break
phone_index = phone_index + 1

8
app/celery/process_ses_receipts_tasks.py
View File

@@ -12,7 +12,7 @@ from app.celery.service_callback_tasks import (
send_complaint_to_service,
send_delivery_status_to_service,
)
from app.config import QueueNames
from app.config import Config, QueueNames
from app.dao import notifications_dao
from app.dao.complaint_dao import save_complaint
from app.dao.notifications_dao import dao_get_notification_history_by_reference
@@ -65,7 +65,9 @@ def process_ses_results(self, response):
f"Callback may have arrived before notification was"
f"persisted to the DB. Adding task to retry queue"
)
self.retry(queue=QueueNames.RETRY)
self.retry(
queue=QueueNames.RETRY, expires=Config.DEFAULT_REDIS_EXPIRE_TIME
)
else:
current_app.logger.warning(
f"Notification not found for reference: {reference} "
@@ -115,7 +117,7 @@ def process_ses_results(self, response):
except Exception:
current_app.logger.exception("Error processing SES results")
self.retry(queue=QueueNames.RETRY)
self.retry(queue=QueueNames.RETRY, expires=Config.DEFAULT_REDIS_EXPIRE_TIME)
def determine_notification_bounce_type(ses_message):

24
app/celery/provider_tasks.py
View File

@@ -10,7 +10,7 @@ from app import aws_cloudwatch_client, notify_celery, redis_store
from app.clients.email import EmailClientNonRetryableException
from app.clients.email.aws_ses import AwsSesClientThrottlingSendRateException
from app.clients.sms import SmsClientResponseException
from app.config import QueueNames
from app.config import Config, QueueNames
from app.dao import notifications_dao
from app.dao.notifications_dao import (
sanitize_successful_notification_by_id,
@@ -128,6 +128,8 @@ def deliver_sms(self, notification_id):
)
# Code branches off to send_to_providers.py
message_id = send_to_providers.send_sms_to_provider(notification)
# DEPRECATED
# We have to put it in UTC. For other timezones, the delay
# will be ignored and it will fire immediately (although this probably only affects developer testing)
my_eta = utc_now() + timedelta(seconds=DELIVERY_RECEIPT_DELAY_IN_SECONDS)
@@ -152,9 +154,15 @@ def deliver_sms(self, notification_id):
try:
if self.request.retries == 0:
self.retry(queue=QueueNames.RETRY, countdown=0)
self.retry(
queue=QueueNames.RETRY,
countdown=0,
expires=Config.DEFAULT_REDIS_EXPIRE_TIME,
)
else:
self.retry(queue=QueueNames.RETRY)
self.retry(
queue=QueueNames.RETRY, expires=Config.DEFAULT_REDIS_EXPIRE_TIME
)
except self.MaxRetriesExceededError:
message = (
"RETRY FAILED: Max retries reached. The task send_sms_to_provider failed for notification {}. "
@@ -170,7 +178,7 @@ def deliver_sms(self, notification_id):
@notify_celery.task(
bind=True, name="deliver_email", max_retries=48, default_retry_delay=300
bind=True, name="deliver_email", max_retries=48, default_retry_delay=30
)
def deliver_email(self, notification_id):
try:
@@ -182,8 +190,12 @@ def deliver_email(self, notification_id):
if not notification:
raise NoResultFound()
personalisation = redis_store.get(f"email-personalisation-{notification_id}")
recipient = redis_store.get(f"email-recipient-{notification_id}")
if personalisation:
notification.personalisation = json.loads(personalisation)
if recipient:
notification.recipient = json.loads(recipient)
notification.personalisation = json.loads(personalisation)
send_to_providers.send_email_to_provider(notification)
except EmailClientNonRetryableException:
current_app.logger.exception(f"Email notification {notification_id} failed")
@@ -199,7 +211,7 @@ def deliver_email(self, notification_id):
f"RETRY: Email notification {notification_id} failed"
)
self.retry(queue=QueueNames.RETRY)
self.retry(queue=QueueNames.RETRY, expires=Config.DEFAULT_REDIS_EXPIRE_TIME)
except self.MaxRetriesExceededError:
message = (
"RETRY FAILED: Max retries reached. "

52
app/celery/tasks.py
View File

@@ -7,7 +7,7 @@ from sqlalchemy.exc import IntegrityError, SQLAlchemyError
from app import create_uuid, encryption, notify_celery
from app.aws import s3
from app.celery import provider_tasks
from app.config import QueueNames
from app.config import Config, QueueNames
from app.dao.inbound_sms_dao import dao_get_inbound_sms_by_id
from app.dao.jobs_dao import dao_get_job_by_id, dao_update_job
from app.dao.notifications_dao import (
@@ -20,7 +20,10 @@ from app.dao.service_sms_sender_dao import dao_get_service_sms_senders_by_id
from app.dao.templates_dao import dao_get_template_by_id
from app.enums import JobStatus, KeyType, NotificationType
from app.errors import TotalRequestsError
from app.notifications.process_notifications import persist_notification
from app.notifications.process_notifications import (
get_notification,
persist_notification,
)
from app.notifications.validators import check_service_over_total_message_limit
from app.serialised_models import SerialisedService, SerialisedTemplate
from app.service.utils import service_allowed_to_send_to
@@ -143,6 +146,7 @@ def process_row(row, template, job, service, sender_id=None):
),
task_kwargs,
queue=QueueNames.DATABASE,
expires=Config.DEFAULT_REDIS_EXPIRE_TIME,
)
return notification_id
@@ -166,7 +170,7 @@ def __total_sending_limits_for_job_exceeded(service, job, job_id):
return True
@notify_celery.task(bind=True, name="save-sms", max_retries=5, default_retry_delay=300)
@notify_celery.task(bind=True, name="save-sms", max_retries=2, default_retry_delay=600)
def save_sms(self, service_id, notification_id, encrypted_notification, sender_id=None):
"""Persist notification to db and place notification in queue to send to sns."""
notification = encryption.decrypt(encrypted_notification)
@@ -271,7 +275,7 @@ def save_email(
"Email {} failed as restricted service".format(notification_id)
)
return
original_notification = get_notification(notification_id)
try:
saved_notification = persist_notification(
template_id=notification["template"],
@@ -288,10 +292,11 @@ def save_email(
notification_id=notification_id,
reply_to_text=reply_to_text,
)
provider_tasks.deliver_email.apply_async(
[str(saved_notification.id)], queue=QueueNames.SEND_EMAIL
)
# we only want to send once
if original_notification is None:
provider_tasks.deliver_email.apply_async(
[str(saved_notification.id)], queue=QueueNames.SEND_EMAIL
)
current_app.logger.debug(
"Email {} created at {}".format(
@@ -310,7 +315,7 @@ def save_api_email(self, encrypted_notification):
@notify_celery.task(
bind=True, name="save-api-sms", max_retries=5, default_retry_delay=300
bind=True, name="save-api-sms", max_retries=2, default_retry_delay=600
)
def save_api_sms(self, encrypted_notification):
save_api_email_or_sms(self, encrypted_notification)
@@ -329,6 +334,8 @@ def save_api_email_or_sms(self, encrypted_notification):
if notification["notification_type"] == NotificationType.EMAIL
else provider_tasks.deliver_sms
)
original_notification = get_notification(notification["id"])
try:
persist_notification(
notification_id=notification["id"],
@@ -346,19 +353,24 @@ def save_api_email_or_sms(self, encrypted_notification):
status=notification["status"],
document_download_count=notification["document_download_count"],
)
# Only get here if save to the db was successful (i.e. first time)
if original_notification is None:
provider_task.apply_async([notification["id"]], queue=q)
current_app.logger.debug(
f"{notification['id']} has been persisted and sent to delivery queue."
)
provider_task.apply_async([notification["id"]], queue=q)
current_app.logger.debug(
f"{notification['notification_type']} {notification['id']} has been persisted and sent to delivery queue."
)
except IntegrityError:
current_app.logger.info(
current_app.logger.warning(
f"{notification['notification_type']} {notification['id']} already exists."
)
# If we don't have the return statement here, we will fall through and end
# up retrying because IntegrityError is a subclass of SQLAlchemyError
return
except SQLAlchemyError:
try:
self.retry(queue=QueueNames.RETRY)
self.retry(queue=QueueNames.RETRY, expires=Config.DEFAULT_REDIS_EXPIRE_TIME)
except self.MaxRetriesExceededError:
current_app.logger.exception(
f"Max retry failed Failed to persist notification {notification['id']}",
@@ -379,7 +391,11 @@ def handle_exception(task, notification, notification_id, exc):
# This probably (hopefully) is not an issue with Redis as the celery backing store
current_app.logger.exception("Retry" + retry_msg)
try:
task.retry(queue=QueueNames.RETRY, exc=exc)
task.retry(
queue=QueueNames.RETRY,
exc=exc,
expires=Config.DEFAULT_REDIS_EXPIRE_TIME,
)
except task.MaxRetriesExceededError:
current_app.logger.exception("Max retry failed" + retry_msg)
@@ -428,7 +444,9 @@ def send_inbound_sms_to_service(self, inbound_sms_id, service_id):
)
if not isinstance(e, HTTPError) or e.response.status_code >= 500:
try:
self.retry(queue=QueueNames.RETRY)
self.retry(
queue=QueueNames.RETRY, expires=Config.DEFAULT_REDIS_EXPIRE_TIME
)
except self.MaxRetriesExceededError:
current_app.logger.exception(
"Retry: send_inbound_sms_to_service has retried the max number of"

1
app/config.py
View File

@@ -53,6 +53,7 @@ class TaskNames(object):
class Config(object):
NOTIFY_APP_NAME = "api"
DEFAULT_REDIS_EXPIRE_TIME = 4 * 24 * 60 * 60
NOTIFY_ENVIRONMENT = getenv("NOTIFY_ENVIRONMENT", "development")
# URL of admin app
ADMIN_BASE_URL = getenv("ADMIN_BASE_URL", "http://localhost:6012")

28
app/dao/api_key_dao.py
View File

@@ -36,13 +36,9 @@ def expire_api_key(service_id, api_key_id):
def get_model_api_keys(service_id, id=None):
if id:
return (
db.session.execute(
select(ApiKey).filter_by(id=id, service_id=service_id, expiry_date=None)
)
.scalars()
.one()
)
return db.session.execute(
select(ApiKey).where(id=id, service_id=service_id, expiry_date=None)
).one()
seven_days_ago = utc_now() - timedelta(days=7)
return (
db.session.execute(
@@ -63,13 +59,9 @@ def get_unsigned_secrets(service_id):
"""
This method can only be exposed to the Authentication of the api calls.
"""
api_keys = (
db.session.execute(
select(ApiKey).filter_by(service_id=service_id, expiry_date=None)
)
.scalars()
.all()
)
api_keys = db.session.execute(
select(ApiKey).where(service_id=service_id, expiry_date=None)
).all()
keys = [x.secret for x in api_keys]
return keys
@@ -78,9 +70,7 @@ def get_unsigned_secret(key_id):
"""
This method can only be exposed to the Authentication of the api calls.
"""
api_key = (
db.session.execute(select(ApiKey).filter_by(id=key_id, expiry_date=None))
.scalars()
.one()
)
api_key = db.session.execute(
select(ApiKey).where(id=key_id, expiry_date=None)
).one()
return api_key.secret

17
app/dao/notifications_dao.py
View File

@@ -72,7 +72,12 @@ def dao_create_notification(notification):
# notify-api-742 remove phone numbers from db
notification.to = "1"
notification.normalised_to = "1"
db.session.add(notification)
# notify-api-1454 insert only if it doesn't exist
stmt = select(Notification).where(Notification.id == notification.id)
result = db.session.execute(stmt).scalar()
if result is None:
db.session.add(notification)
def country_records_delivery(phone_prefix):
@@ -106,6 +111,16 @@ def _update_notification_status(
return notification
def update_notification_message_id(notification_id, message_id):
stmt = (
update(Notification)
.where(Notification.id == notification_id)
.values(message_id=message_id)
)
db.session.execute(stmt)
db.session.commit()
@autocommit
def update_notification_status_by_id(
notification_id, status, sent_by=None, provider_response=None, carrier=None

6
app/delivery/send_to_providers.py
View File

@@ -16,7 +16,10 @@ from app import (
from app.aws.s3 import get_personalisation_from_s3, get_phone_number_from_s3
from app.celery.test_key_tasks import send_email_response, send_sms_response
from app.dao.email_branding_dao import dao_get_email_branding_by_id
from app.dao.notifications_dao import dao_update_notification
from app.dao.notifications_dao import (
dao_update_notification,
update_notification_message_id,
)
from app.dao.provider_details_dao import get_provider_details_by_notification_type
from app.dao.service_sms_sender_dao import dao_get_sms_senders_by_service_id
from app.enums import BrandType, KeyType, NotificationStatus, NotificationType
@@ -117,6 +120,7 @@ def send_sms_to_provider(notification):
message_id = provider.send_sms(**send_sms_kwargs)
current_app.logger.info(f"got message_id {message_id}")
update_notification_message_id(notification.id, message_id)
except Exception as e:
n = notification
msg = f"FAILED send to sms, job_id: {n.job_id} row_number {n.job_row_number} message_id {message_id}"

1
app/models.py
View File

@@ -1532,6 +1532,7 @@ class Notification(db.Model):
provider_response = db.Column(db.Text, nullable=True)
carrier = db.Column(db.Text, nullable=True)
message_id = db.Column(db.Text, nullable=True)
# queue_name = db.Column(db.Text, nullable=True)

5
app/notifications/process_notifications.py
View File

@@ -8,6 +8,7 @@ from app.config import QueueNames
from app.dao.notifications_dao import (
dao_create_notification,
dao_delete_notifications_by_id,
get_notification_by_id,
)
from app.enums import KeyType, NotificationStatus, NotificationType
from app.errors import BadRequestError
@@ -53,6 +54,10 @@ def check_placeholders(template_object):
raise BadRequestError(fields=[{"template": message}], message=message)
def get_notification(notification_id):
return get_notification_by_id(notification_id)
def persist_notification(
*,
template_id,

12
app/service/sender.py
View File

@@ -1,5 +1,8 @@
import json
from flask import current_app
from app import redis_store
from app.config import QueueNames
from app.dao.services_dao import (
dao_fetch_active_users_for_service,
@@ -40,6 +43,15 @@ def send_notification_to_service_users(
key_type=KeyType.NORMAL,
reply_to_text=notify_service.get_default_reply_to_email_address(),
)
redis_store.set(
f"email-personalisation-{notification.id}",
json.dumps(personalisation),
ex=24 * 60 * 60,
)
redis_store.set(
f"email-recipient-{notification.id}", notification.to, ex=24 * 60 * 60
)
send_notification_to_queue(notification, queue=QueueNames.NOTIFY)