<commit_before>from datetime import datetime, timedelta

import iso8601
from celery.exceptions import Retry
from flask import current_app, json
from notifications_utils.statsd_decorators import statsd
from sqlalchemy.orm.exc import NoResultFound

from app import notify_celery, statsd_client
from app.config import QueueNames
from app.clients.email.aws_ses import get_aws_responses
from app.dao import notifications_dao
from app.models import NOTIFICATION_SENDING, NOTIFICATION_PENDING

from app.notifications.notifications_ses_callback import (
    determine_notification_bounce_type,
    handle_complaint,
    _check_and_queue_complaint_callback_task,
    _check_and_queue_callback_task,
)


@notify_celery.task(bind=True, name="process-ses-result", max_retries=5, default_retry_delay=300)
@statsd(namespace="tasks")
def process_ses_results(self, response):
    try:
        ses_message = json.loads(response['Message'])
        notification_type = ses_message['notificationType']

        if notification_type == 'Bounce':
            notification_type = determine_notification_bounce_type(notification_type, ses_message)
        elif notification_type == 'Complaint':
            _check_and_queue_complaint_callback_task(*handle_complaint(ses_message))
            return True

        aws_response_dict = get_aws_responses(notification_type)

        notification_status = aws_response_dict['notification_status']
        reference = ses_message['mail']['messageId']

        try:
            notification = notifications_dao.dao_get_notification_or_history_by_reference(reference=reference)
        except NoResultFound:
            message_time = iso8601.parse_date(ses_message['mail']['timestamp']).replace(tzinfo=None)
            if datetime.utcnow() - message_time < timedelta(minutes=5):
                self.retry(queue=QueueNames.RETRY)
            else:
                current_app.logger.warning(
                    "notification not found for reference: {} (update to {})".format(reference, notification_status)
                )
            return

        if notification.status not in [NOTIFICATION_SENDING, NOTIFICATION_PENDING]:
            notifications_dao._duplicate_update_warning(
                notification=notification,
                status=notification_status
            )
            return
        else:
            notifications_dao.dao_update_notifications_by_reference(
                references=[reference],
                update_dict={'status': notification_status}
            )
<commit_msg>Add better logging to understand callback delivery cases<commit_after>from datetime import datetime, timedelta

import iso8601
from celery.exceptions import Retry
from flask import current_app, json
from notifications_utils.statsd_decorators import statsd
from sqlalchemy.orm.exc import NoResultFound

from app import notify_celery, statsd_client
from app.config import QueueNames
from app.clients.email.aws_ses import get_aws_responses
from app.dao import notifications_dao
from app.models import NOTIFICATION_SENDING, NOTIFICATION_PENDING

from app.notifications.notifications_ses_callback import (
    determine_notification_bounce_type,
    handle_complaint,
    _check_and_queue_complaint_callback_task,
    _check_and_queue_callback_task,
)


@notify_celery.task(bind=True, name="process-ses-result", max_retries=5, default_retry_delay=300)
@statsd(namespace="tasks")
def process_ses_results(self, response):
    try:
        ses_message = json.loads(response['Message'])
        notification_type = ses_message['notificationType']

        if notification_type == 'Bounce':
            notification_type = determine_notification_bounce_type(notification_type, ses_message)
        elif notification_type == 'Complaint':
            _check_and_queue_complaint_callback_task(*handle_complaint(ses_message))
            return True

        aws_response_dict = get_aws_responses(notification_type)

        notification_status = aws_response_dict['notification_status']
        reference = ses_message['mail']['messageId']

        try:
            notification = notifications_dao.dao_get_notification_or_history_by_reference(reference=reference)
        except NoResultFound:
            message_time = iso8601.parse_date(ses_message['mail']['timestamp']).replace(tzinfo=None)
            if datetime.utcnow() - message_time < timedelta(minutes=5):
                current_app.logger.info(
                    f"notification not found for reference: {reference} (update to {notification_status}). "
                    f"Callback may have arrived before notification was persisted to the DB. Adding task to retry queue"
                )
                self.retry(queue=QueueNames.RETRY)
            else:
                current_app.logger.warning(
                    f"notification not found for reference: {reference} (update to {notification_status})"
                )
            return

        if notification.status not in [NOTIFICATION_SENDING, NOTIFICATION_PENDING]:
            notifications_dao._duplicate_update_warning(
                notification=notification,
                status=notification_status
            )
            return
        else:
            notifications_dao.dao_update_notifications_by_reference(
                references=[reference],
                update_dict={'status': notification_status}
            )

{'fraction_non_alphanumeric': 0.06912527798369163, 'fraction_numerical': 0.004818383988139363, 'mean_word_length': 3.276545166402536, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5886919', 'n_tokens_mistral': 1460, 'n_tokens_neox': 1406, 'n_words': 279}