SQL Repeat Record Report & Case Data Report

corehq/apps/data_interfaces/tasks.py

@@ -20,9 +20,11 @@
 )
 from corehq.form_processor.utils.general import should_use_sql_backend
 from corehq.motech.repeaters.dbaccessors import (
-    get_repeat_records_by_payload_id,
-    iter_repeat_records_by_repeater,
+    get_couch_repeat_record_ids_by_payload_id,
+    get_sql_repeat_records_by_payload_id,
+    iter_repeat_record_ids_by_repeater,
 )
+from corehq.motech.repeaters.models import SQLRepeatRecord
 from corehq.sql_db.util import get_db_aliases_for_partitioned_query
 from corehq.toggles import DISABLE_CASE_UPDATE_RULE_SCHEDULED_TASK
 from corehq.util.decorators import serial_task
@@ -224,8 +226,13 @@ def delete_old_rule_submission_logs():
 
 
 @task(serializer='pickle')
-def task_operate_on_payloads(record_ids, domain, action=''):
-    return operate_on_payloads(record_ids, domain, action,
+def task_operate_on_payloads(
+    record_ids: List[str],
+    domain: str,
+    action,  # type: Literal['resend', 'cancel', 'requeue']  # 3.8+
+    use_sql: bool,
+):
+    return operate_on_payloads(record_ids, domain, action, use_sql,
                                task=task_operate_on_payloads)
 
 
@@ -234,24 +241,35 @@ def task_generate_ids_and_operate_on_payloads(
     payload_id: Optional[str],
     repeater_id: Optional[str],
     domain: str,
-    action: str = '',
+    action,  # type: Literal['resend', 'cancel', 'requeue']  # 3.8+
+    use_sql: bool,
 ) -> dict:
-    repeat_record_ids = _get_repeat_record_ids(payload_id, repeater_id, domain)
-    return operate_on_payloads(repeat_record_ids, domain, action,
+    repeat_record_ids = _get_repeat_record_ids(payload_id, repeater_id, domain,
+                                               use_sql)
+    return operate_on_payloads(repeat_record_ids, domain, action, use_sql,
                                task=task_generate_ids_and_operate_on_payloads)
 
 
 def _get_repeat_record_ids(
     payload_id: Optional[str],
     repeater_id: Optional[str],
     domain: str,
+    use_sql: bool,
 ) -> List[str]:
     if not payload_id and not repeater_id:
         return []
     if payload_id:
-        results = get_repeat_records_by_payload_id(domain, payload_id)
+        if use_sql:
+            records = get_sql_repeat_records_by_payload_id(domain, payload_id)
+            return [r.id for r in records]
+        else:
+            return get_couch_repeat_record_ids_by_payload_id(domain, payload_id)
     else:
-        results = iter_repeat_records_by_repeater(domain, repeater_id)
-    ids = [x['id'] for x in results]
-
-    return ids
+        if use_sql:
+            queryset = SQLRepeatRecord.objects.filter(
+                domain=domain,
+                repeater_stub__repeater_id=repeater_id,
+            )
+            return [r['id'] for r in queryset.values('id')]
+        else:
+            return list(iter_repeat_record_ids_by_repeater(domain, repeater_id))

corehq/apps/data_interfaces/tests/test_tasks.py

@@ -9,45 +9,77 @@
 
 class TestTasks(TestCase):
 
-    def test_task_operate_on_payloads_no_action(self):
+    @patch('corehq.apps.data_interfaces.utils.DownloadBase')
+    @patch('corehq.apps.data_interfaces.utils._get_couch_repeat_record')
+    def test_task_operate_on_payloads_no_action(
+        self,
+        unused_1,
+        unused_2,
+    ):
         response = task_operate_on_payloads(
             record_ids=['payload_id'],
             domain='test_domain',
-            action=''
+            action='',
+            use_sql=False,
         )
-        self.assertEqual(response,
-                         {'messages': {'errors': ['No action specified']}})
+        self.assertEqual(response, {
+            'messages': {
+                'errors': [
+                    "Could not perform action for repeat record (id=payload_id): "
+                    "Unknown action ''",
+                ],
+                'success': [],
+                'success_count_msg': '',
+            }
+        })
 
     def test_task_operate_on_payloads_no_payload_ids(self):
         response = task_operate_on_payloads(
             record_ids=[],
             domain='test_domain',
-            action='test_action'
+            action='test_action',
+            use_sql=False,
         )
         self.assertEqual(response,
                          {'messages': {'errors': ['No payloads specified']}})
 
+    @patch('corehq.apps.data_interfaces.utils.DownloadBase')
+    @patch('corehq.apps.data_interfaces.utils._get_couch_repeat_record')
     @patch('corehq.apps.data_interfaces.tasks._get_repeat_record_ids')
     def test_task_generate_ids_and_operate_on_payloads_no_action(
         self,
         get_repeat_record_ids_mock,
+        unused_1,
+        unused_2,
     ):
         get_repeat_record_ids_mock.return_value = ['c0ffee', 'deadbeef']
         response = task_generate_ids_and_operate_on_payloads(
             payload_id='c0ffee',
             repeater_id=None,
             domain='test_domain',
-            action=''
+            action='',
+            use_sql=False,
         )
-        self.assertEqual(response,
-                         {'messages': {'errors': ['No action specified']}})
+        self.assertEqual(response, {
+            'messages': {
+                'errors': [
+                    "Could not perform action for repeat record (id=c0ffee): "
+                    "Unknown action ''",
+                    "Could not perform action for repeat record (id=deadbeef): "
+                    "Unknown action ''",
+                ],
+                'success': [],
+                'success_count_msg': '',
+            }
+        })
 
     def test_task_generate_ids_and_operate_on_payloads_no_data(self):
         response = task_generate_ids_and_operate_on_payloads(
             payload_id=None,
             repeater_id=None,
             domain='test_domain',
-            action=''
+            action='',
+            use_sql=False,
         )
         self.assertEqual(response,
                          {'messages': {'errors': ['No payloads specified']}})