Merge pull request #4731 from alphagov/high-failure-rates-task-replica

Move "check_for_services_with_high_failure_rates_or_sending_to_tv_numbers" task to read replica

Author: klssmith

app/celery/scheduled_tasks.py

@@ -456,8 +456,16 @@ def check_for_services_with_high_failure_rates_or_sending_to_tv_numbers():
     end_date = datetime.utcnow()
     message = ""
 
-    services_with_failures = dao_find_services_with_high_failure_rates(start_date=start_date, end_date=end_date)
-    services_sending_to_tv_numbers = dao_find_services_sending_to_tv_numbers(start_date=start_date, end_date=end_date)
+    services_with_failures = dao_find_services_with_high_failure_rates(
+        start_date=start_date,
+        end_date=end_date,
+        retry_attempts=2,
+    )
+    services_sending_to_tv_numbers = dao_find_services_sending_to_tv_numbers(
+        start_date=start_date,
+        end_date=end_date,
+        retry_attempts=2,
+    )
 
     if services_with_failures:
         message += (

app/dao/services_dao.py

@@ -494,9 +494,12 @@ def dao_fetch_active_users_for_service(service_id):
     return query.all()
 
 
-def dao_find_services_sending_to_tv_numbers(start_date, end_date, threshold=500):
+@retryable_query()
+def dao_find_services_sending_to_tv_numbers(
+    start_date: datetime, end_date: datetime, threshold: int = 500, session: Session | scoped_session = db.session
+):
     return (
-        db.session.query(
+        session.query(
             Notification.service_id.label("service_id"), func.count(Notification.id).label("notification_count")
         )
         .filter(
@@ -517,9 +520,12 @@ def dao_find_services_sending_to_tv_numbers(start_date, end_date, threshold=500)
     )
 
 
-def dao_find_services_with_high_failure_rates(start_date, end_date, threshold=10000):
+@retryable_query()
+def dao_find_services_with_high_failure_rates(
+    start_date: datetime, end_date: datetime, threshold: int = 10000, session: Session | scoped_session = db.session
+):
     subquery = (
-        db.session.query(func.count(Notification.id).label("total_count"), Notification.service_id.label("service_id"))
+        session.query(func.count(Notification.id).label("total_count"), Notification.service_id.label("service_id"))
         .filter(
             Notification.service_id == Service.id,
             Notification.created_at >= start_date,
@@ -538,7 +544,7 @@ def dao_find_services_with_high_failure_rates(start_date, end_date, threshold=10
     subquery = subquery.subquery()
 
     query = (
-        db.session.query(
+        session.query(
             Notification.service_id.label("service_id"),
             func.count(Notification.id).label("permanent_failure_count"),
             subquery.c.total_count.label("total_count"),

tests/app/dao/test_services_dao.py

@@ -1189,7 +1189,15 @@ def create_email_sms_letter_template():
 
 
 @freeze_time("2019-12-02 12:00:00.000000")
-def test_dao_find_services_sending_to_tv_numbers(notify_db_session, fake_uuid):
+@pytest.mark.parametrize(
+    "session,expected_bind_key",
+    (
+        (db.session, None),
+        (db.session_bulk, "bulk"),
+    ),
+    ids=("default", "bulk"),
+)
+def test_dao_find_services_sending_to_tv_numbers(notify_db_session, fake_uuid, session, expected_bind_key):
     service_1 = create_service(service_name="Service 1", service_id=fake_uuid)
     service_3 = create_service(service_name="Service 3", restricted=True)  # restricted is excluded
     service_4 = create_service(service_name="Service 4", active=False)  # not active is excluded
@@ -1224,12 +1232,23 @@ def test_dao_find_services_sending_to_tv_numbers(notify_db_session, fake_uuid):
     start_date = datetime.utcnow() - timedelta(days=1)
     end_date = datetime.utcnow()
 
-    result = dao_find_services_sending_to_tv_numbers(start_date, end_date, threshold=4)
+    with QueryRecorder() as query_recorder:
+        result = dao_find_services_sending_to_tv_numbers(start_date, end_date, threshold=4, session=session)
+
+    assert {query_info.bind_key for query_info in query_recorder.queries} == {expected_bind_key}
     assert len(result) == 1
     assert str(result[0].service_id) == fake_uuid
 
 
-def test_dao_find_services_with_high_failure_rates(notify_db_session, fake_uuid):
+@pytest.mark.parametrize(
+    "session,expected_bind_key",
+    (
+        (db.session, None),
+        (db.session_bulk, "bulk"),
+    ),
+    ids=("default", "bulk"),
+)
+def test_dao_find_services_with_high_failure_rates(notify_db_session, fake_uuid, session, expected_bind_key):
     service_1 = create_service(service_name="Service 1", service_id=fake_uuid)
     service_3 = create_service(service_name="Service 3", restricted=True)  # restricted is excluded
     service_4 = create_service(service_name="Service 4", active=False)  # not active is excluded
@@ -1258,9 +1277,10 @@ def test_dao_find_services_with_high_failure_rates(notify_db_session, fake_uuid)
     start_date = datetime.utcnow() - timedelta(days=1)
     end_date = datetime.utcnow()
 
-    result = dao_find_services_with_high_failure_rates(start_date, end_date, threshold=3)
-    # assert len(result) == 3
-    # assert str(result[0].service_id) == fake_uuid
+    with QueryRecorder() as query_recorder:
+        result = dao_find_services_with_high_failure_rates(start_date, end_date, threshold=3, session=session)
+
+    assert {query_info.bind_key for query_info in query_recorder.queries} == {expected_bind_key}
     assert len(result) == 1
     assert str(result[0].service_id) == fake_uuid
     assert result[0].permanent_failure_rate == 0.25