Function to delete high volume notifications for a service

This command will allow us to explore the timing of deleting notifications, which is why it's written as a command. The intention is to try this strategy out then lift and shift the code into a task.

The next commit will do the insert into NotificationHistory and drop the back up table.

app/commands.py

@@ -11,7 +11,7 @@ from click_datetime import Datetime as click_dt
 from flask import current_app, json
 from notifications_utils.recipients import RecipientCSV
 from notifications_utils.template import SMSMessageTemplate
-from sqlalchemy.exc import IntegrityError
+from sqlalchemy.exc import IntegrityError, SQLAlchemyError
 from sqlalchemy.orm.exc import NoResultFound
 from notifications_utils.statsd_decorators import statsd
 
@@ -45,7 +45,6 @@ from app.dao.templates_dao import dao_get_template_by_id
 from app.dao.users_dao import delete_model_user, delete_user_verify_codes, get_user_by_email
 from app.models import (
     PROVIDERS,
-    NOTIFICATION_CREATED,
     KEY_TYPE_TEST,
     SMS_TYPE,
     EMAIL_TYPE,
@@ -57,6 +56,10 @@ from app.models import (
     Service,
     EmailBranding,
     LetterBranding,
+    NOTIFICATION_CREATED,
+    NOTIFICATION_DELIVERED,
+    NOTIFICATION_PERMANENT_FAILURE,
+    NOTIFICATION_TEMPORARY_FAILURE,
 )
 from app.performance_platform.processing_time import send_processing_time_for_start_and_end
 from app.utils import get_london_midnight_in_utc, get_midnight_for_day_before
@@ -901,3 +904,105 @@ def process_row_from_job(job_id, job_row_number):
             notification_id = process_row(row, template, job, job.service)
             current_app.logger.info("Process row {} for job {} created notification_id: {}".format(
                 job_row_number, job_id, notification_id))
+
+
+@notify_command(name='delete-high-volume-service-data')
+@click.option('-i', '--service_id', required=True, help='Service id of the high volume service')
+@click.option('-s', '--start_date', required=True, type=click_dt(format='%Y-%m-%d %H'),
+              help='Start date of process YYYY-MM-DD HH:mm')
+@click.option('-e', '--end_date', required=True, type=click_dt(format='%Y-%m-%d %H'),
+              help='End date of process YYYY-MM-DD HH:MM')
+@click.option('-t', '--notification_type', required=False, default='email',
+              help='Notification type of the data to delete')
+def delete_high_volume_service_data(service_id, start_date, end_date, notification_type):
+    str_date = start_date.strftime('%Y_%m_%d_%H')
+    str_end_date = end_date.strftime('%Y_%m_%d_%H')
+
+    bkup_tble_name = f'back_up_notifications_{str_date}_to_{str_end_date}'
+
+    print(f"""Creating back up {bkup_tble_name} starting at: {datetime.utcnow()}
+          for {notification_type} notifications for service: {service_id}, starting at {start_date} and {end_date}
+         """)
+
+    _create_notification_bkup_table(bkup_tble_name, end_date, notification_type, service_id, start_date)
+
+    hour_start = start_date
+    hour_end = hour_start + timedelta(hours=1)
+
+    terminate_statuses = [NOTIFICATION_DELIVERED, NOTIFICATION_TEMPORARY_FAILURE, NOTIFICATION_PERMANENT_FAILURE]
+    delete_query = Notification.query.filter(
+        Notification.notification_type == notification_type,
+        Notification.service_id == service_id,
+        Notification.created_at >= hour_start,
+        Notification.created_at <= hour_end,
+        Notification.status.in_(terminate_statuses)
+    )
+    # Iterate hour by hour
+    del_count = 0
+    while hour_start < end_date:
+        del_count += delete_query.delete(synchronize_session=False)
+        db.session.commit()
+        # print(hour_start, hour_end)
+        # increment hour
+        hour_end = hour_end + timedelta(hours=1)
+        hour_start = hour_start + timedelta(hours=1)
+        delete_query = Notification.query.filter(
+            Notification.notification_type == notification_type,
+            Notification.service_id == service_id,
+            Notification.created_at >= hour_start,
+            Notification.created_at <= hour_end,
+            Notification.status.in_(terminate_statuses)
+        )
+
+    print(f"""Completed deleting {del_count} from notifications
+              for {notification_type} notifications for
+              service: {service_id}, starting at {start_date} and {end_date}
+             """)
+
+
+def _create_notification_bkup_table(bkup_tble_name, end_date, notification_type, service_id, start_date):
+    try:
+        create_tbl_sql = f"""
+            CREATE TABLE {bkup_tble_name} AS 
+                SELECT *
+                  FROM notifications
+                 WHERE service_id = :service_id
+                   AND notification_type = :notification_type
+                   AND created_at >= :start_date
+                   AND created_at <= :end_date
+                   AND key_type = 'normal'
+                   AND notification_status in ('delivered', 'permanent-failure', 'temporary-failure')
+        """
+        input_params = {
+            "service_id": service_id,
+            "notification_type": notification_type,
+            "start_date": start_date,
+            "end_date": end_date
+        }
+        db.session.execute(create_tbl_sql, input_params)
+        db.session.commit()
+    except SQLAlchemyError as e:
+        db.session.commit()  # terminate previous transaction
+        # This query isn't quite right yet, if the notifications are already deleted still get 0.
+        # however it doesn't cause any harm, because there is nothing to delete.
+        # But it will also return 0 if the rows exist in notifications
+        qry = f""" Select count(*) from notifications 
+          WHERE service_id = :service_id
+          AND notification_type = :notification_type
+          AND created_at >= :start_date
+          AND created_at <= :end_date
+          AND key_type = 'normal'
+          UNION 
+          SELECT count(*) 
+            FROM {bkup_tble_name} 
+        """
+        result = db.session.execute(qry, input_params).fetchall()
+        if result[0][0] == result[1][0]:
+            print("Table and data already exists, keep going")
+            return
+        else:
+            # This will throw the exception if the data is already deleted or partically deleted...
+            # but gives us a chance to see what's happending.
+            print(f"Table already exists but row counts are inconsistent is missing bail out. "
+                  f"There are {result[0][0]} rows in notifications and {result[1][0]} rows in {bkup_tble_name}")
+            raise e

app/dao/inbound_sms_dao.py

@@ -102,7 +102,7 @@ def _delete_inbound_sms(datetime_to_delete_from, query_filter):
     while number_deleted > 0:
         _insert_inbound_sms_history(subquery, query_limit=query_limit)
 
-        number_deleted = InboundSms.query.filter(InboundSms.id.in_(subquery)).delete(synchronize_session=False)
+        number_deleted = InboundSms.query.filter(InboundSms.id.in_(subquery)).delete(synchronize_session='fetch')
         deleted += number_deleted
 
     return deleted

app/dao/notifications_dao.py

@@ -356,7 +356,7 @@ def _move_notifications_to_notification_history(notification_type, service_id, d
                 )
             )
 
-        insert_update_notification_history(notification_type, timestamp_to_delete_backwards_from, service_id)
+        insert_update_notification_history(notification_type, timestamp_to_delete_backwards_from, service_id, qry_limit)
 
         if service_id == '539d63a1-701d-400d-ab11-f3ee2319d4d4':
             current_app.logger.info(
@@ -366,24 +366,53 @@ def _move_notifications_to_notification_history(notification_type, service_id, d
             )
 
         deleted += _delete_notifications(
-            notification_type, timestamp_to_delete_backwards_from, service_id
+            notification_type, timestamp_to_delete_backwards_from, service_id, qry_limit
         )
 
     return deleted
 
 
-def _delete_notifications(notification_type, date_to_delete_from, service_id):
-    deleted = Notification.query.filter(
+def _delete_notifications(notification_type, date_to_delete_from, service_id, query_limit):
+    subquery = db.session.query(
+        Notification.id
+    ).join(NotificationHistory, NotificationHistory.id == Notification.id).filter(
         Notification.notification_type == notification_type,
         Notification.service_id == service_id,
         Notification.created_at < date_to_delete_from,
-    ).delete(synchronize_session=False)
-    db.session.commit()
+    ).limit(query_limit).subquery()
+
+    deleted = _delete_for_query(subquery)
+
+    subquery_for_test_keys = db.session.query(
+        Notification.id
+    ).filter(
+        Notification.notification_type == notification_type,
+        Notification.service_id == service_id,
+        Notification.created_at < date_to_delete_from,
+        Notification.key_type == KEY_TYPE_TEST
+    ).limit(query_limit).subquery()
+
+    deleted += _delete_for_query(subquery_for_test_keys)
 
     return deleted
 
 
-def insert_update_notification_history(notification_type, date_to_delete_from, service_id):
+def _delete_for_query(subquery):
+    number_deleted = db.session.query(Notification).filter(
+        Notification.id.in_(subquery)).delete(synchronize_session='fetch')
+    deleted = number_deleted
+    db.session.commit()
+    while number_deleted > 0:
+        number_deleted = db.session.query(Notification).filter(
+            Notification.id.in_(subquery)).delete(synchronize_session='fetch')
+        deleted += number_deleted
+        db.session.commit()
+    return deleted
+
+
+def insert_update_notification_history(notification_type, date_to_delete_from, service_id, query_limit=10000):
+    offset = 0
+
     notification_query = db.session.query(
         *[x.name for x in NotificationHistory.__table__.c]
     ).filter(
@@ -391,26 +420,32 @@ def insert_update_notification_history(notification_type, date_to_delete_from, s
         Notification.service_id == service_id,
         Notification.created_at < date_to_delete_from,
         Notification.key_type != KEY_TYPE_TEST
+    ).order_by(
+        Notification.created_at
     )
+    notifications_count = notification_query.count()
 
-    stmt = insert(NotificationHistory).from_select(
-        NotificationHistory.__table__.c,
-        notification_query
-    )
+    while offset < notifications_count:
+        stmt = insert(NotificationHistory).from_select(
+            NotificationHistory.__table__.c,
+            notification_query.limit(query_limit).offset(offset)
+        )
 
-    stmt = stmt.on_conflict_do_update(
-        constraint="notification_history_pkey",
-        set_={
-            "notification_status": stmt.excluded.status,
-            "reference": stmt.excluded.reference,
-            "billable_units": stmt.excluded.billable_units,
-            "updated_at": stmt.excluded.updated_at,
-            "sent_at": stmt.excluded.sent_at,
-            "sent_by": stmt.excluded.sent_by
-        }
-    )
-    db.session.connection().execute(stmt)
-    db.session.commit()
+        stmt = stmt.on_conflict_do_update(
+            constraint="notification_history_pkey",
+            set_={
+                "notification_status": stmt.excluded.status,
+                "reference": stmt.excluded.reference,
+                "billable_units": stmt.excluded.billable_units,
+                "updated_at": stmt.excluded.updated_at,
+                "sent_at": stmt.excluded.sent_at,
+                "sent_by": stmt.excluded.sent_by
+            }
+        )
+        db.session.connection().execute(stmt)
+        db.session.commit()
+
+        offset += query_limit
 
 
 def _delete_letters_from_s3(

tests/app/dao/notification_dao/test_notification_dao_delete_notifications.py

@@ -9,13 +9,13 @@ from freezegun import freeze_time
 
 from app.dao.notifications_dao import (
     delete_notifications_older_than_retention_by_type,
+    db,
     insert_update_notification_history
 )
 from app.models import Notification, NotificationHistory
 from tests.app.db import (
     create_template,
     create_notification,
-    create_notification_history,
     create_service_data_retention,
     create_service
 )
@@ -75,7 +75,7 @@ def test_should_delete_notifications_by_type_after_seven_days(
 ):
     mocker.patch("app.dao.notifications_dao.get_s3_bucket_objects")
     email_template, letter_template, sms_template = _create_templates(sample_service)
-    # create one notification a day between 1st and 10th from 01:00 to 11:00 of each type
+    # create one notification a day between 1st and 10th from 11:00 to 19:00 of each type
     for i in range(1, 11):
         past_date = '2016-0{0}-{1:02d}  {1:02d}:00:00.000000'.format(month, i)
         with freeze_time(past_date):
@@ -157,24 +157,17 @@ def test_delete_notifications_inserts_notification_history(sample_service):
     assert NotificationHistory.query.count() == 2
 
 
-def test_delete_notifications_updates_notification_history(notify_db, sample_email_template, mocker):
+def test_delete_notifications_updates_notification_history(sample_email_template, mocker):
     mocker.patch("app.dao.notifications_dao.get_s3_bucket_objects")
-    now = datetime.utcnow()
-    notification = create_notification(
-        template=sample_email_template,
-        created_at=datetime.utcnow() - timedelta(days=8),
-        reference="ses_reference",
-        billable_units=1,
-        updated_at=now,
-        sent_by="ses",
-        status='delivered'
-    )
-
-    create_notification_history(
-        id=notification.id,
-        template=sample_email_template,
-        created_at=datetime.utcnow() - timedelta(days=8),
-        status='sending',
+    notification = create_notification(template=sample_email_template, created_at=datetime.utcnow() - timedelta(days=8))
+    Notification.query.filter_by(id=notification.id).update(
+        {"status": "delivered",
+         "reference": "ses_reference",
+         "billable_units": 1,  # I know we don't update this for emails but this is a unit test
+         "updated_at": datetime.utcnow(),
+         "sent_at": datetime.utcnow(),
+         "sent_by": "ses"
+         }
     )
 
     delete_notifications_older_than_retention_by_type("email")
@@ -184,7 +177,7 @@ def test_delete_notifications_updates_notification_history(notify_db, sample_ema
     assert history[0].status == 'delivered'
     assert history[0].reference == 'ses_reference'
     assert history[0].billable_units == 1
-    assert history[0].updated_at == now
+    assert history[0].updated_at
     assert history[0].sent_by == 'ses'
 
 
@@ -234,6 +227,21 @@ def test_delete_notifications_does_try_to_delete_from_s3_when_letter_has_not_bee
     mock_get_s3.assert_not_called()
 
 
+@freeze_time("2016-01-10 12:00:00.000000")
+def test_should_not_delete_notification_if_history_does_not_exist(sample_service, mocker):
+    mocker.patch("app.dao.notifications_dao.get_s3_bucket_objects")
+    mocker.patch("app.dao.notifications_dao.insert_update_notification_history")
+    with freeze_time('2016-01-01 12:00'):
+        email_template, letter_template, sms_template = _create_templates(sample_service)
+        create_notification(template=email_template, status='permanent-failure')
+        create_notification(template=sms_template, status='delivered')
+        create_notification(template=letter_template, status='temporary-failure')
+    assert Notification.query.count() == 3
+    delete_notifications_older_than_retention_by_type('sms')
+    assert Notification.query.count() == 3
+    assert NotificationHistory.query.count() == 0
+
+
 def test_delete_notifications_calls_subquery_multiple_times(sample_template):
     create_notification(template=sample_template, created_at=datetime.now() - timedelta(days=8))
     create_notification(template=sample_template, created_at=datetime.now() - timedelta(days=8))
@@ -280,6 +288,35 @@ def test_insert_update_notification_history(sample_service):
     assert notification_3.id in history_ids
 
 
+def test_insert_update_notification_history_with_more_notifications_than_query_limit(mocker, sample_service):
+    template = create_template(sample_service, template_type='sms')
+    notification_1 = create_notification(template=template, created_at=datetime.utcnow() - timedelta(days=3))
+    notification_2 = create_notification(template=template, created_at=datetime.utcnow() - timedelta(days=8))
+    notification_3 = create_notification(template=template, created_at=datetime.utcnow() - timedelta(days=9))
+    other_types = ['email', 'letter']
+    for template_type in other_types:
+        t = create_template(service=sample_service, template_type=template_type)
+        create_notification(template=t, created_at=datetime.utcnow() - timedelta(days=3))
+        create_notification(template=t, created_at=datetime.utcnow() - timedelta(days=8))
+
+    db_connection_spy = mocker.spy(db.session, 'connection')
+    db_commit_spy = mocker.spy(db.session, 'commit')
+
+    insert_update_notification_history(
+        notification_type='sms', date_to_delete_from=datetime.utcnow() - timedelta(days=7),
+        service_id=sample_service.id, query_limit=1)
+    history = NotificationHistory.query.all()
+    assert len(history) == 2
+
+    history_ids = [x.id for x in history]
+    assert notification_1.id not in history_ids
+    assert notification_2.id in history_ids
+    assert notification_3.id in history_ids
+
+    assert db_connection_spy.call_count == 2
+    assert db_commit_spy.call_count == 2
+
+
 def test_insert_update_notification_history_only_insert_update_given_service(sample_service):
     other_service = create_service(service_name='another service')
     other_template = create_template(service=other_service)

tests/app/db.py

@@ -324,8 +324,7 @@ def create_notification_history(
         international=False,
         phone_prefix=None,
         created_by_id=None,
-        postage=None,
-        id=None
+        postage=None
 ):
     assert job or template
     if job:
@@ -342,7 +341,7 @@ def create_notification_history(
         postage = 'second'
 
     data = {
-        'id': id or uuid.uuid4(),
+        'id': uuid.uuid4(),
         'job_id': job and job.id,
         'job': job,
         'service_id': template.service.id,