Latest and greatest.

Signed-off-by: Cliff Hill <Clifford.hill@gsa.gov>

app/celery/process_ses_receipts_tasks.py

@@ -1,7 +1,6 @@
 from datetime import timedelta
 
 import iso8601
-from celery.exceptions import Retry
 from flask import current_app, json
 from sqlalchemy.orm.exc import NoResultFound
 
@@ -26,7 +25,12 @@ from app.utils import utc_now
 
 
 @notify_celery.task(
-    bind=True, name="process-ses-result", max_retries=5, default_retry_delay=300
+    bind=True,
+    name="process-ses-result",
+    autoretry_for=(Exception,),
+    # throws=(Exception,),  # Been attempted, did nothing.
+    max_retries=5,
+    default_retry_delay=300,
 )
 def process_ses_results(self, response):
     try:
@@ -65,7 +69,7 @@ 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)
+                raise
             else:
                 current_app.logger.warning(
                     f"Notification not found for reference: {reference} "
@@ -95,27 +99,25 @@ def process_ses_results(self, response):
 
         if not aws_response_dict["success"]:
             current_app.logger.info(
-                "SES delivery failed: notification id {} and reference {} has error found. Status {}".format(
+                f"SES delivery failed: notification id {notification.id} and reference "
-                    notification.id, reference, aws_response_dict["message"]
+                f"{reference} has error found. Status {aws_response_dict['message']}"
-                )
             )
         else:
             current_app.logger.info(
-                "SES callback return status of {} for notification: {}".format(
+                f"SES callback return status of {notification_status} "
-                    notification_status, notification.id
+                f"for notification: {notification.id}"
-                )
             )
 
         check_and_queue_callback_task(notification)
 
         return True
 
-    except Retry:
+    except Exception as e:
-        raise
+        print("Exception REACHED")
-
+        print(type(e))
-    except Exception:
+        print(e)
         current_app.logger.exception("Error processing SES results")
-        self.retry(queue=QueueNames.RETRY)
+        raise
 
 
 def determine_notification_bounce_type(ses_message):

app/celery/provider_tasks.py

@@ -1,6 +1,8 @@
 import json
+import logging
 import os
 from datetime import timedelta
+from functools import wraps
 
 from botocore.exceptions import ClientError
 from flask import current_app
@@ -31,6 +33,7 @@ DELIVERY_RECEIPT_DELAY_IN_SECONDS = 30
     name="check_sms_delivery_receipt",
     max_retries=48,
     default_retry_delay=300,
+    autoretry_for=(NotificationTechnicalFailureException, ClientError),
 )
 def check_sms_delivery_receipt(self, message_id, notification_id, sent_at):
     """
@@ -52,7 +55,7 @@ def check_sms_delivery_receipt(self, message_id, notification_id, sent_at):
             status, provider_response, carrier = aws_cloudwatch_client.check_sms(
                 message_id, notification_id, sent_at
             )
-        except NotificationTechnicalFailureException as ntfe:
+        except NotificationTechnicalFailureException:
             provider_response = "Unable to find carrier response -- still looking"
             status = "pending"
             carrier = ""
@@ -62,7 +65,7 @@ def check_sms_delivery_receipt(self, message_id, notification_id, sent_at):
                 carrier=carrier,
                 provider_response=provider_response,
             )
-            raise self.retry(exc=ntfe)
+            raise
         except ClientError as err:
             # Probably a ThrottlingException but could be something else
             error_code = err.response["Error"]["Code"]
@@ -77,7 +80,7 @@ def check_sms_delivery_receipt(self, message_id, notification_id, sent_at):
                 carrier=carrier,
                 provider_response=provider_response,
             )
-            raise self.retry(exc=err)
+            raise
 
     if status == "success":
         status = NotificationStatus.DELIVERED
@@ -104,8 +107,40 @@ def check_sms_delivery_receipt(self, message_id, notification_id, sent_at):
         )
 
 
+def _deliver_sms_task_handler(cls):
+    """Handle the max retries exceeded error case for delivering sms notifications."""
+
+    func = cls.__call__
+
+    @wraps(func)
+    def deliver_sms_task_wrapper(self, notification_id):
+        try:
+            return func(self, notification_id)
+        except self.MaxRetriesExceededError:
+            message = (
+                "RETRY FAILED: Max retries reached. The task send_sms_to_provider failed for notification {}. "
+                "Notification has been updated to technical-failure".format(
+                    notification_id
+                )
+            )
+            update_notification_status_by_id(
+                notification_id,
+                NotificationStatus.TECHNICAL_FAILURE,
+            )
+            raise NotificationTechnicalFailureException(message)
+
+    cls.__call__ = deliver_sms_task_wrapper
+
+    return cls
+
+
+@_deliver_sms_task_handler
 @notify_celery.task(
-    bind=True, name="deliver_sms", max_retries=48, default_retry_delay=300
+    bind=True,
+    name="deliver_sms",
+    max_retries=48,
+    default_retry_delay=300,
+    autoretry_for=(Exception,),
 )
 def deliver_sms(self, notification_id):
     """Branch off to the final step in delivering the notification to sns and get delivery receipts."""
@@ -141,26 +176,37 @@ def deliver_sms(self, notification_id):
             notification_id,
             NotificationStatus.TEMPORARY_FAILURE,
         )
+
         if isinstance(e, SmsClientResponseException):
-            current_app.logger.warning(
+            log_lvl = logging.WARNING
-                "SMS notification delivery for id: {} failed".format(notification_id),
+            log_exc_info = False
-            )
         else:
-            current_app.logger.exception(
+            log_lvl = logging.ERROR
-                "SMS notification delivery for id: {} failed".format(notification_id),
+            log_exc_info = True
+
+        current_app.logger.log(
+            level=log_lvl,
+            msg=f"SMS notification delivery for id: {notification_id} failed",
+            exc_info=log_exc_info,
         )
 
+        raise
+
+
+def _deliver_email_task_handler(cls):
+    """Handle the max retries exceeded error case for delivering email notifications."""
+
+    func = cls.__call__
+
+    @wraps(func)
+    def deliver_email_task_wrapper(self, notification_id):
         try:
-            if self.request.retries == 0:
+            return func(self, notification_id)
-                self.retry(queue=QueueNames.RETRY, countdown=0)
-            else:
-                self.retry(queue=QueueNames.RETRY)
         except self.MaxRetriesExceededError:
             message = (
-                "RETRY FAILED: Max retries reached. The task send_sms_to_provider failed for notification {}. "
+                "RETRY FAILED: Max retries reached. "
-                "Notification has been updated to technical-failure".format(
+                f"The task send_email_to_provider failed for notification {notification_id}. "
-                    notification_id
+                "Notification has been updated to technical-failure"
-                )
             )
             update_notification_status_by_id(
                 notification_id,
@@ -168,9 +214,19 @@ def deliver_sms(self, notification_id):
             )
             raise NotificationTechnicalFailureException(message)
 
+    cls.__call__ = deliver_email_task_wrapper
 
+    return cls
+
+
+@_deliver_email_task_handler
 @notify_celery.task(
-    bind=True, name="deliver_email", max_retries=48, default_retry_delay=30
+    bind=True,
+    name="deliver_email",
+    max_retries=48,
+    default_retry_delay=30,
+    autoretry_for=(Exception,),
+    dont_autoretry_for=(EmailClientNonRetryableException,),
 )
 def deliver_email(self, notification_id):
     try:
@@ -191,9 +247,12 @@ def deliver_email(self, notification_id):
         send_to_providers.send_email_to_provider(notification)
     except EmailClientNonRetryableException:
         current_app.logger.exception(f"Email notification {notification_id} failed")
-        update_notification_status_by_id(notification_id, "technical-failure")
+        update_notification_status_by_id(
+            notification_id,
+            NotificationStatus.TECHNICAL_FAILURE,
+        )
+        raise
     except Exception as e:
-        try:
         if isinstance(e, AwsSesClientThrottlingSendRateException):
             current_app.logger.warning(
                 f"RETRY: Email notification {notification_id} was rate limited by SES"
@@ -203,17 +262,4 @@ def deliver_email(self, notification_id):
                 f"RETRY: Email notification {notification_id} failed"
             )
 
-            self.retry(queue=QueueNames.RETRY)
+        raise
-        except self.MaxRetriesExceededError:
-            message = (
-                "RETRY FAILED: Max retries reached. "
-                "The task send_email_to_provider failed for notification {}. "
-                "Notification has been updated to technical-failure".format(
-                    notification_id
-                )
-            )
-            update_notification_status_by_id(
-                notification_id,
-                NotificationStatus.TECHNICAL_FAILURE,
-            )
-            raise NotificationTechnicalFailureException(message)

app/celery/service_callback_tasks.py

@@ -1,15 +1,68 @@
 import json
+from functools import wraps
+from inspect import signature
 
 from flask import current_app
 from requests import HTTPError, RequestException, request
 
 from app import encryption, notify_celery
-from app.config import QueueNames
 from app.utils import DATETIME_FORMAT
 
 
+def _send_to_service_task_handler(cls):
+
+    func = cls.__call__
+
+    @wraps(func)
+    def send_to_service_task_wrapper(*args, **kwargs):
+        sig = signature(func)
+        bargs = sig.bind(*args, **kwargs)
+        bargs.apply_defaults()
+
+        function_name = func.__name__
+
+        if function_name == "send_delivery_status_to_service":
+            encrypted_status_update = bargs.arguments["encrypted_status_update"]
+
+            status_update = encryption.decrypt(encrypted_status_update)
+            service_callback_url = status_update["service_callback_api_url"]
+
+            notification_id = bargs.arguments["notification_id"]
+
+        elif function_name == "send_complaint_to_service":
+            complaint_data = bargs.arguments["complaint_data"]
+
+            notification_id = complaint_data["notification_id"]
+            service_callback_url = complaint_data["service_callback_api_url"]
+
+        else:
+            raise ValueError(
+                f"Incorrect send to service function name found: {function_name}"
+            )
+
+        self_ = bargs.arguments["self"]
+
+        try:
+            return func(*args, **kwargs)
+        except self_.MaxRetriesExceededError:
+            current_app.logger.warning(
+                f"Retry: {function_name} has retried the max num of times for callback url "
+                f"{service_callback_url} and notification_id: {notification_id}"
+            )
+            raise
+
+    cls.__call__ = send_to_service_task_wrapper
+
+    return cls
+
+
+@_send_to_service_task_handler
 @notify_celery.task(
-    bind=True, name="send-delivery-status", max_retries=5, default_retry_delay=300
+    bind=True,
+    name="send-delivery-status",
+    max_retries=5,
+    default_retry_delay=300,
+    autoretry_for=(HTTPError,),
 )
 def send_delivery_status_to_service(self, notification_id, encrypted_status_update):
     status_update = encryption.decrypt(encrypted_status_update)
@@ -36,8 +89,13 @@ def send_delivery_status_to_service(self, notification_id, encrypted_status_upda
     )
 
 
+@_send_to_service_task_handler
 @notify_celery.task(
-    bind=True, name="send-complaint", max_retries=5, default_retry_delay=300
+    bind=True,
+    name="send-complaint",
+    max_retries=5,
+    default_retry_delay=300,
+    autoretry_for=(HTTPError,),
 )
 def send_complaint_to_service(self, complaint_data):
     complaint = encryption.decrypt(complaint_data)
@@ -72,43 +130,29 @@ def _send_data_to_service_callback_api(
             data=json.dumps(data),
             headers={
                 "Content-Type": "application/json",
-                "Authorization": "Bearer {}".format(token),
+                "Authorization": f"Bearer {token}",
             },
             timeout=5,
         )
         current_app.logger.info(
-            "{} sending {} to {}, response {}".format(
+            f"{function_name} sending {notification_id} to {service_callback_url}, response {response.status_code}"
-                function_name,
-                notification_id,
-                service_callback_url,
-                response.status_code,
-            )
         )
         response.raise_for_status()
     except RequestException as e:
         current_app.logger.warning(
-            "{} request failed for notification_id: {} and url: {}. exception: {}".format(
+            f"{function_name} request failed for notification_id: {notification_id} and "
-                function_name, notification_id, service_callback_url, e
+            f"url: {service_callback_url}. exception: {e}"
-            )
         )
         if (
             not isinstance(e, HTTPError)
             or e.response.status_code >= 500
             or e.response.status_code == 429
         ):
-            try:
+            raise
-                self.retry(queue=QueueNames.CALLBACKS_RETRY)
-            except self.MaxRetriesExceededError:
-                current_app.logger.warning(
-                    "Retry: {} has retried the max num of times for callback url {} and notification_id: {}".format(
-                        function_name, service_callback_url, notification_id
-                    )
-                )
         else:
             current_app.logger.warning(
-                "{} callback is not being retried for notification_id: {} and url: {}. exception: {}".format(
+                f"{function_name} callback is not being retried for notification_id: "
-                    function_name, notification_id, service_callback_url, e
+                f"{notification_id} and url: {service_callback_url}. exception: {e}"
-                )
             )
 
 

app/celery/tasks.py

@@ -1,4 +1,6 @@
 import json
+from functools import wraps
+from inspect import signature
 
 from flask import current_app
 from requests import HTTPError, RequestException, request
@@ -166,7 +168,38 @@ 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)
+def _save_task_hander(func):
+    @wraps(func)
+    def save_task_wrapper(*args, **kwargs):
+        sig = signature(func)
+        bargs = sig.bind(*args, **kwargs)
+        bargs.apply_defaults()
+
+        task = bargs.arguments["self"]
+        notification_id = bargs.arguments["notification_id"]
+        notification = encryption.decrypt(bargs.arguments["encrypted_notification"])
+
+        try:
+            return func(*args, **kwargs)
+        except task.MaxRetriesExceededError:
+            retry_msg = (
+                f"{task.__name__} notification for job {notification.get("job", None)} "
+            )
+            f"row number {notification.get("row_number", None)} and notification id {notification_id}"
+            current_app.logger.exception("Max retry failed" + retry_msg)
+            raise
+
+    return save_task_wrapper
+
+
+@_save_task_hander
+@notify_celery.task(
+    bind=True,
+    name="save-sms",
+    max_retries=5,
+    default_retry_delay=300,
+    autoretry_for=(SQLAlchemyError,),
+)
 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)
@@ -194,9 +227,7 @@ def save_sms(self, service_id, notification_id, encrypted_notification, sender_i
                 f"service not allowed to send for job_id {notification.get('job', None)}, aborting"
             )
         )
-        current_app.logger.debug(
+        current_app.logger.debug(f"SMS {notification_id} failed as restricted service")
-            "SMS {} failed as restricted service".format(notification_id)
-        )
         return
 
     try:
@@ -235,19 +266,27 @@ def save_sms(self, service_id, notification_id, encrypted_notification, sender_i
         )
 
         current_app.logger.debug(
-            "SMS {} created at {} for job {}".format(
+            f"SMS {saved_notification.id} created at {saved_notification.created_at} for job "
-                saved_notification.id,
+            f"{notification.get('job', None)}"
-                saved_notification.created_at,
-                notification.get("job", None),
-            )
         )
 
-    except SQLAlchemyError as e:
+    except SQLAlchemyError:
-        handle_exception(self, notification, notification_id, e)
+        if not get_notification_by_id(notification_id):
+            retry_msg = (
+                f"{self.__name__} notification for job {notification.get("job", None)} "
+            )
+            f"row number {notification.get("row_number", None)} and notification id {notification_id}"
+            current_app.logger.exception(retry_msg)
+            raise
 
 
+@_save_task_hander
 @notify_celery.task(
-    bind=True, name="save-email", max_retries=5, default_retry_delay=300
+    bind=True,
+    name="save-email",
+    max_retries=5,
+    default_retry_delay=300,
+    autoretry_for=(SQLAlchemyError,),
 )
 def save_email(
     self, service_id, notification_id, encrypted_notification, sender_id=None
@@ -267,9 +306,7 @@ def save_email(
         reply_to_text = template.reply_to_text
 
     if not service_allowed_to_send_to(notification["to"], service, KeyType.NORMAL):
-        current_app.logger.info(
+        current_app.logger.info(f"Email {notification_id} failed as restricted service")
-            "Email {} failed as restricted service".format(notification_id)
-        )
         return
 
     try:
@@ -294,23 +331,57 @@ def save_email(
         )
 
         current_app.logger.debug(
-            "Email {} created at {}".format(
+            f"Email {saved_notification.id} created at {saved_notification.created_at}"
-                saved_notification.id, saved_notification.created_at
         )
+    except SQLAlchemyError:
+        if not get_notification_by_id(notification_id):
+            retry_msg = (
+                f"{self.__name__} notification for job {notification.get("job", None)} "
+                f"row number {notification.get("row_number", None)} and notification id {notification_id}"
             )
-    except SQLAlchemyError as e:
+            current_app.logger.exception(retry_msg)
-        handle_exception(self, notification, notification_id, e)
+            raise
 
 
+def _save_api_task_handler(func):
+    @wraps(func)
+    def save_api_task_wrapper(*args, **kwargs):
+        sig = signature(func)
+        bargs = sig.bind(*args, **kwargs)
+        bargs.apply_defaults()
+
+        self_ = bargs.argument["self"]
+        notification = encryption.decrypt[bargs.arguments["encrypted_notification"]]
+        try:
+            return func(*args, **kwargs)
+        except self_.MaxRetriesExceededError:
+            current_app.logger.exception(
+                f"Max retry failed Failed to persist notification {notification['id']}",
+            )
+            raise
+
+    return save_api_task_wrapper
+
+
+@_save_api_task_handler
 @notify_celery.task(
-    bind=True, name="save-api-email", max_retries=5, default_retry_delay=300
+    bind=True,
+    name="save-api-email",
+    max_retries=5,
+    default_retry_delay=300,
+    autoretry_for=(SQLAlchemyError,),
 )
 def save_api_email(self, encrypted_notification):
     save_api_email_or_sms(self, encrypted_notification)
 
 
+@_save_api_task_handler
 @notify_celery.task(
-    bind=True, name="save-api-sms", max_retries=5, default_retry_delay=300
+    bind=True,
+    name="save-api-sms",
+    max_retries=5,
+    default_retry_delay=300,
+    autoretry_for=(SQLAlchemyError,),
 )
 def save_api_sms(self, encrypted_notification):
     save_api_email_or_sms(self, encrypted_notification)
@@ -360,36 +431,29 @@ def save_api_email_or_sms(self, encrypted_notification):
         # up retrying because IntegrityError is a subclass of SQLAlchemyError
         return
 
-    except SQLAlchemyError:
+
+def _send_inbound_sms_to_service_handler(func):
+    @wraps(func)
+    def send_inbound_sms_to_service_wrapper(self, inbound_sms_id, service_id):
         try:
-            self.retry(queue=QueueNames.RETRY)
+            return func(self, inbound_sms_id, service_id)
         except self.MaxRetriesExceededError:
             current_app.logger.exception(
-                f"Max retry failed Failed to persist notification {notification['id']}",
+                "Retry: send_inbound_sms_to_service has retried the max number of"
+                + f"times for service: {service_id} and inbound_sms {inbound_sms_id}"
             )
+            raise
+
+    return send_inbound_sms_to_service_wrapper
 
 
-def handle_exception(task, notification, notification_id, exc):
+@_send_inbound_sms_to_service_handler
-    if not get_notification_by_id(notification_id):
-        retry_msg = "{task} notification for job {job} row number {row} and notification id {noti}".format(
-            task=task.__name__,
-            job=notification.get("job", None),
-            row=notification.get("row_number", None),
-            noti=notification_id,
-        )
-        # Sometimes, SQS plays the same message twice. We should be able to catch an IntegrityError, but it seems
-        # SQLAlchemy is throwing a FlushError. So we check if the notification id already exists then do not
-        # send to the retry queue.
-        # 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)
-        except task.MaxRetriesExceededError:
-            current_app.logger.exception("Max retry failed" + retry_msg)
-
-
 @notify_celery.task(
-    bind=True, name="send-inbound-sms", max_retries=5, default_retry_delay=300
+    bind=True,
+    name="send-inbound-sms",
+    max_retries=5,
+    default_retry_delay=300,
+    autoretry_for=(RequestException,),
 )
 def send_inbound_sms_to_service(self, inbound_sms_id, service_id):
     inbound_api = get_service_inbound_api_for_service(service_id=service_id)
@@ -431,13 +495,7 @@ def send_inbound_sms_to_service(self, inbound_sms_id, service_id):
             + f"and url: {inbound_api.url}. exception: {e}"
         )
         if not isinstance(e, HTTPError) or e.response.status_code >= 500:
-            try:
+            raise
-                self.retry(queue=QueueNames.RETRY)
-            except self.MaxRetriesExceededError:
-                current_app.logger.exception(
-                    "Retry: send_inbound_sms_to_service has retried the max number of"
-                    + f"times for service: {service_id} and inbound_sms {inbound_sms_id}"
-                )
         else:
             current_app.logger.warning(
                 f"send_inbound_sms_to_service is not being retried for service_id: {service_id} for "

tests/app/celery/test_process_ses_receipts_tasks.py

@@ -1,6 +1,7 @@
 import json
 from unittest.mock import ANY
 
+import pytest
 from freezegun import freeze_time
 
 from app import encryption
@@ -157,6 +158,9 @@ def test_process_ses_results_retry_called(sample_email_template, mocker):
     mocked = mocker.patch(
         "app.celery.process_ses_receipts_tasks.process_ses_results.retry"
     )
+    with pytest.raises(Exception):  # noqa: B017
+        # In order to make this work, we have to suppress the flake8 warning about
+        # pytest.raises(Exception), which is usually considered a bad thing.
         process_ses_results(response=ses_notification_callback(reference="ref1"))
     assert mocked.call_count != 0
 
@@ -240,17 +244,29 @@ def test_ses_callback_should_not_update_notification_status_if_already_delivered
     assert mock_upd.call_count == 0
 
 
-def test_ses_callback_should_retry_if_notification_is_new(mocker):
+def test_ses_callback_should_retry_if_notification_is_new(client, _notify_db, mocker):
-    mock_retry = mocker.patch(
+    # mock_retry = mocker.patch(
-        "app.celery.process_ses_receipts_tasks.process_ses_results.retry"
+    #     "app.celery.process_ses_receipts_tasks.process_ses_results.retry"
-    )
+    # )
     mock_logger = mocker.patch(
-        "app.celery.process_ses_receipts_tasks.current_app.logger.error"
+        "app.celery.process_ses_receipts_tasks.current_app.logger.exception"
     )
     with freeze_time("2017-11-17T12:14:03.646Z"):
-        assert process_ses_results(ses_notification_callback(reference="ref")) is None
+        try:
+            assert (
+                process_ses_results(ses_notification_callback(reference="ref")) is None
+            )
+        except Exception as e:
+            import traceback
+            print(type(e))
+            print("*" * 80)
+            print(e)
+            print("-" * 80)
+            print(traceback.format_exc())
+            print("-" * 80)
+            raise
         assert mock_logger.call_count == 0
-        assert mock_retry.call_count == 1
+        # assert mock_retry.call_count == 1
 
 
 def test_ses_callback_should_log_if_notification_is_missing(client, _notify_db, mocker):