Merge branch 'master' into notifications-get-api-key-filter

app/celery/provider_tasks.py

@@ -1,9 +1,8 @@
-import json
-
 from datetime import datetime
 from monotonic import monotonic
 from flask import current_app
-from app import notify_celery, statsd_client, encryption, clients
+from app import notify_celery, statsd_client, clients, create_uuid
+from app.clients.email import EmailClientException
 from app.clients.sms import SmsClientException
 from app.dao.notifications_dao import (
     update_provider_stats,
@@ -11,19 +10,17 @@ from app.dao.notifications_dao import (
     dao_update_notification, update_notification_status_by_id)
 from app.dao.provider_details_dao import get_provider_details_by_notification_type
 from app.dao.services_dao import dao_fetch_service_by_id
-from app.celery.research_mode_tasks import send_sms_response
-
+from app.celery.research_mode_tasks import send_sms_response, send_email_response
 from notifications_utils.recipients import (
     validate_and_format_phone_number
 )
 
 from app.dao.templates_dao import dao_get_template_by_id
 from notifications_utils.template import (
-    Template,
-    unlink_govuk_escaped
+    Template
 )
 
-from app.models import SMS_TYPE
+from app.models import SMS_TYPE, EMAIL_TYPE
 
 
 def retry_iteration_to_delay(retry=0):
@@ -50,7 +47,7 @@ def retry_iteration_to_delay(retry=0):
 
 
 @notify_celery.task(bind=True, name="send-sms-to-provider", max_retries=5, default_retry_delay=5)
-def send_sms_to_provider(self, service_id, notification_id, encrypted_notification=None):
+def send_sms_to_provider(self, service_id, notification_id):
     task_start = monotonic()
 
     service = dao_fetch_service_by_id(service_id)
@@ -98,7 +95,7 @@ def send_sms_to_provider(self, service_id, notification_id, encrypted_notificati
                 update_notification_status_by_id(notification.id, 'technical-failure', 'failure')
 
         current_app.logger.info(
-            "SMS {} created at {} sent at {}".format(notification_id, notification.created_at, notification.sent_at)
+            "SMS {} sent to provider at {}".format(notification_id, notification.sent_at)
         )
         statsd_client.incr("notifications.tasks.send-sms-to-provider")
         statsd_client.timing("notifications.tasks.send-sms-to-provider.task-time", monotonic() - task_start)
@@ -117,3 +114,61 @@ def provider_to_use(notification_type, notification_id):
         raise Exception("No active {} providers".format(notification_type))
 
     return clients.get_client_by_name_and_type(active_providers_in_order[0].identifier, notification_type)
+
+
+@notify_celery.task(bind=True, name="send-email-to-provider", max_retries=5, default_retry_delay=5)
+def send_email_to_provider(self, service_id, notification_id, reply_to_addresses=None):
+    task_start = monotonic()
+    service = dao_fetch_service_by_id(service_id)
+    provider = provider_to_use(EMAIL_TYPE, notification_id)
+    notification = get_notification_by_id(notification_id)
+    if notification.status == 'created':
+        try:
+            template = Template(
+                dao_get_template_by_id(notification.template_id, notification.template_version).__dict__,
+                values=notification.personalisation
+            )
+
+            if service.research_mode:
+                reference = str(create_uuid())
+                send_email_response.apply_async(
+                    (provider.get_name(), reference, notification.to), queue='research-mode'
+                )
+            else:
+                from_address = '"{}" <{}@{}>'.format(service.name, service.email_from,
+                                                     current_app.config['NOTIFY_EMAIL_DOMAIN'])
+                reference = provider.send_email(
+                    from_address,
+                    notification.to,
+                    template.replaced_subject,
+                    body=template.replaced_govuk_escaped,
+                    html_body=template.as_HTML_email,
+                    reply_to_addresses=reply_to_addresses,
+                )
+
+                update_provider_stats(
+                    notification_id,
+                    EMAIL_TYPE,
+                    provider.get_name()
+                )
+            notification.reference = reference
+            notification.sent_at = datetime.utcnow()
+            notification.sent_by = provider.get_name(),
+            notification.status = 'sending'
+            dao_update_notification(notification)
+        except EmailClientException as e:
+            try:
+                current_app.logger.error(
+                    "Email notification {} failed".format(notification_id)
+                )
+                current_app.logger.exception(e)
+                self.retry(queue="retry", countdown=retry_iteration_to_delay(self.request.retries))
+            except self.MaxRetriesExceededError:
+                update_notification_status_by_id(notification.id, 'technical-failure', 'failure')
+
+        current_app.logger.info(
+            "Email {} sent to provider at {}".format(notification_id, notification.sent_at)
+        )
+        statsd_client.incr("notifications.tasks.send-email-to-provider")
+        statsd_client.timing("notifications.tasks.send-email-to-provider.task-time", monotonic() - task_start)
+        statsd_client.timing("notifications.email.total-time", datetime.utcnow() - notification.created_at)

app/celery/tasks.py

@@ -19,18 +19,14 @@ from app import (
     encryption
 )
 from app.aws import s3
-from app.celery.provider_tasks import send_sms_to_provider
-from app.celery.research_mode_tasks import send_email_response
-from app.clients.email import EmailClientException
+from app.celery.provider_tasks import send_sms_to_provider, send_email_to_provider
 from app.dao.jobs_dao import (
     dao_update_job,
     dao_get_job_by_id
 )
 from app.dao.notifications_dao import (
     dao_create_notification,
-    dao_update_notification,
-    dao_get_notification_statistics_for_service_and_day,
-    update_provider_stats
+    dao_get_notification_statistics_for_service_and_day
 )
 from app.dao.provider_details_dao import get_provider_details_by_notification_type
 from app.dao.services_dao import dao_fetch_service_by_id
@@ -152,29 +148,12 @@ def send_sms(self,
         return
 
     try:
-
-        sent_at = datetime.utcnow()
-        notification_db_object = Notification(
-            id=notification_id,
-            template_id=notification['template'],
-            template_version=notification['template_version'],
-            to=notification['to'],
-            service_id=service_id,
-            job_id=notification.get('job', None),
-            job_row_number=notification.get('row_number', None),
-            status='created',
-            created_at=datetime.strptime(created_at, DATETIME_FORMAT),
-            personalisation=notification.get('personalisation'),
-            notification_type=SMS_TYPE,
-            api_key_id=api_key_id,
-            key_type=key_type
-        )
-        dao_create_notification(notification_db_object, SMS_TYPE)
+        _save_notification(created_at, notification, notification_id, service_id, SMS_TYPE, api_key_id, key_type)
 
         send_sms_to_provider.apply_async((service_id, notification_id), queue='sms')
 
         current_app.logger.info(
-            "SMS {} created at {} sent at {}".format(notification_id, created_at, sent_at)
+            "SMS {} created at {}".format(notification_id, created_at)
         )
 
         statsd_client.incr("notifications.tasks.send-sms")
@@ -184,8 +163,8 @@ def send_sms(self,
         raise self.retry(queue="retry", exc=e)
 
 
-@notify_celery.task(name="send-email")
-def send_email(service_id,
+@notify_celery.task(bind=True, name="send-email", max_retries=5, default_retry_delay=5)
+def send_email(self, service_id,
                notification_id,
                encrypted_notification,
                created_at,
@@ -196,85 +175,41 @@ def send_email(service_id,
     notification = encryption.decrypt(encrypted_notification)
     service = dao_fetch_service_by_id(service_id)
 
-    provider = provider_to_use(EMAIL_TYPE, notification_id)
-
     if not service_allowed_to_send_to(notification['to'], service):
-        current_app.logger.info(
-            "Email {} failed as restricted service".format(notification_id)
-        )
+        current_app.logger.info("Email {} failed as restricted service".format(notification_id))
         return
 
     try:
-        sent_at = datetime.utcnow()
-        notification_db_object = Notification(
-            id=notification_id,
-            template_id=notification['template'],
-            template_version=notification['template_version'],
-            to=notification['to'],
-            service_id=service_id,
-            job_id=notification.get('job', None),
-            job_row_number=notification.get('row_number', None),
-            status='sending',
-            created_at=datetime.strptime(created_at, DATETIME_FORMAT),
-            sent_at=sent_at,
-            sent_by=provider.get_name(),
-            personalisation=notification.get('personalisation'),
-            notification_type=EMAIL_TYPE,
-            api_key_id=api_key_id,
-            key_type=key_type
-        )
+        _save_notification(created_at, notification, notification_id, service_id, EMAIL_TYPE, api_key_id, key_type)
 
-        dao_create_notification(notification_db_object, EMAIL_TYPE)
-        statsd_client.timing_with_dates(
-            "notifications.tasks.send-email.queued-for",
-            sent_at,
-            datetime.strptime(created_at, DATETIME_FORMAT)
-        )
+        send_email_to_provider.apply_async((service_id, notification_id, reply_to_addresses),
+                                           queue='email')
 
-        try:
-            template = Template(
-                dao_get_template_by_id(notification['template'], notification['template_version']).__dict__,
-                values=notification.get('personalisation', {})
-            )
-
-            if service.research_mode:
-                reference = create_uuid()
-                send_email_response.apply_async(
-                    (provider.get_name(), str(reference), notification['to']), queue='research-mode'
-                )
-            else:
-                from_address = '"{}" <{}@{}>'.format(service.name, service.email_from,
-                                                     current_app.config['NOTIFY_EMAIL_DOMAIN'])
-                reference = provider.send_email(
-                    from_address,
-                    notification['to'],
-                    template.replaced_subject,
-                    body=template.replaced_govuk_escaped,
-                    html_body=template.as_HTML_email,
-                    reply_to_addresses=reply_to_addresses,
-                )
-
-                update_provider_stats(
-                    notification_id,
-                    'email',
-                    provider.get_name()
-                )
-
-            notification_db_object.reference = reference
-            dao_update_notification(notification_db_object)
-
-        except EmailClientException as e:
-            current_app.logger.exception(e)
-            notification_db_object.status = 'technical-failure'
-            dao_update_notification(notification_db_object)
-
-        current_app.logger.info(
-            "Email {} created at {} sent at {}".format(notification_id, created_at, sent_at)
-        )
+        current_app.logger.info("Email {} created at {}".format(notification_id, created_at))
         statsd_client.incr("notifications.tasks.send-email")
         statsd_client.timing("notifications.tasks.send-email.task-time", monotonic() - task_start)
     except SQLAlchemyError as e:
         current_app.logger.exception(e)
+        raise self.retry(queue="retry", exc=e)
+
+
+def _save_notification(created_at, notification, notification_id, service_id, notification_type, api_key_id, key_type):
+    notification_db_object = Notification(
+        id=notification_id,
+        template_id=notification['template'],
+        template_version=notification['template_version'],
+        to=notification['to'],
+        service_id=service_id,
+        job_id=notification.get('job', None),
+        job_row_number=notification.get('row_number', None),
+        status='created',
+        created_at=datetime.strptime(created_at, DATETIME_FORMAT),
+        personalisation=notification.get('personalisation'),
+        notification_type=EMAIL_TYPE,
+        api_key_id=api_key_id,
+        key_type=key_type
+    )
+    dao_create_notification(notification_db_object, notification_type)
 
 
 def service_allowed_to_send_to(recipient, service):