set delete(synchronize_session=False) on bulk deletes

When we're bulk updating, make sure we call `synchronize_session=False`,
and make sure that we then commit before we try and access any ORM
objects in the session that might be deleted.

Tutorial time:

when you delete, sqlalchemy needs to work out what to do with objects in
the session. "evaluate", "fetch", or False (do nothing). It wants to
remove items from the session if you're about to delete them, so that
you don't get confused about the state of objects

* evaluate compares the delete query to each item in the session in
  turn. this is the default. if you have lots in the session this could
  be super slow I guess but that's rarely the case for us. This can lead
  to errors if the column names are different to the model names, like
  on our notification and history models.
* fetch runs the delete query as if it's a select, and then checks the
  results of that vs the session. This could be slow.
* False doesn't do anything. This means the session will be stale and
  potentially will contain now-deleted items, until we call `commit` or
  `expire_all` on the session.

https://docs.sqlalchemy.org/en/13/orm/query.html?highlight=query.update#sqlalchemy.orm.query.Query.delete

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, SQLAlchemyError
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm.exc import NoResultFound
 from notifications_utils.statsd_decorators import statsd
 
@@ -45,6 +45,7 @@ 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,
@@ -56,10 +57,6 @@ 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
@@ -904,105 +901,3 @@ 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='fetch')
+        number_deleted = InboundSms.query.filter(InboundSms.id.in_(subquery)).delete(synchronize_session=False)
         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, qry_limit)
+        insert_update_notification_history(notification_type, timestamp_to_delete_backwards_from, service_id)
 
         if service_id == '539d63a1-701d-400d-ab11-f3ee2319d4d4':
             current_app.logger.info(
@@ -366,53 +366,24 @@ def _move_notifications_to_notification_history(notification_type, service_id, d
             )
 
         deleted += _delete_notifications(
-            notification_type, timestamp_to_delete_backwards_from, service_id, qry_limit
+            notification_type, timestamp_to_delete_backwards_from, service_id
         )
 
     return deleted
 
 
-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(
+def _delete_notifications(notification_type, date_to_delete_from, service_id):
+    deleted = Notification.query.filter(
         Notification.notification_type == notification_type,
         Notification.service_id == service_id,
         Notification.created_at < date_to_delete_from,
-    ).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 _delete_for_query(subquery):
-    number_deleted = db.session.query(Notification).filter(
-        Notification.id.in_(subquery)).delete(synchronize_session='fetch')
-    deleted = number_deleted
+    ).delete(synchronize_session=False)
     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
-
+def insert_update_notification_history(notification_type, date_to_delete_from, service_id):
     notification_query = db.session.query(
         *[x.name for x in NotificationHistory.__table__.c]
     ).filter(
@@ -420,32 +391,26 @@ 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()
 
-    while offset < notifications_count:
-        stmt = insert(NotificationHistory).from_select(
-            NotificationHistory.__table__.c,
-            notification_query.limit(query_limit).offset(offset)
-        )
+    stmt = insert(NotificationHistory).from_select(
+        NotificationHistory.__table__.c,
+        notification_query
+    )
 
-        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
+    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()
 
 
 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 11:00 to 19:00 of each type
+    # create one notification a day between 1st and 10th from 01:00 to 11: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,17 +157,24 @@ def test_delete_notifications_inserts_notification_history(sample_service):
     assert NotificationHistory.query.count() == 2
 
 
-def test_delete_notifications_updates_notification_history(sample_email_template, mocker):
+def test_delete_notifications_updates_notification_history(notify_db, sample_email_template, mocker):
     mocker.patch("app.dao.notifications_dao.get_s3_bucket_objects")
-    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"
-         }
+    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',
     )
 
     delete_notifications_older_than_retention_by_type("email")
@@ -177,7 +184,7 @@ def test_delete_notifications_updates_notification_history(sample_email_template
     assert history[0].status == 'delivered'
     assert history[0].reference == 'ses_reference'
     assert history[0].billable_units == 1
-    assert history[0].updated_at
+    assert history[0].updated_at == now
     assert history[0].sent_by == 'ses'
 
 
@@ -227,21 +234,6 @@ 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))
@@ -288,35 +280,6 @@ 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,7 +324,8 @@ def create_notification_history(
         international=False,
         phone_prefix=None,
         created_by_id=None,
-        postage=None
+        postage=None,
+        id=None
 ):
     assert job or template
     if job:
@@ -341,7 +342,7 @@ def create_notification_history(
         postage = 'second'
 
     data = {
-        'id': uuid.uuid4(),
+        'id': id or uuid.uuid4(),
         'job_id': job and job.id,
         'job': job,
         'service_id': template.service.id,