Merge pull request #5055 from broadinstitute/trigger-delete-families

Trigger delete families

Author: hanars

clickhouse_search/search.py

@@ -667,11 +667,6 @@ def delete_clickhouse_project(project, dataset_type, sample_type=None):
     return f'Deleted all {dataset_type} search data for project {project.name}'
 
 
-def delete_clickhouse_family(project, family_guid, dataset_type, sample_type=None):
-    dataset_type = _clickhouse_dataset_type(dataset_type, sample_type)
-    return f'Clickhouse does not support deleting individual families from project. Manually delete {dataset_type} data for {family_guid} in project {project.guid}'
-
-
 SV_DATASET_TYPES = {
     Sample.SAMPLE_TYPE_WGS: Sample.DATASET_TYPE_SV_CALLS,
     Sample.SAMPLE_TYPE_WES: 'GCNV',

seqr/management/commands/transfer_families_to_different_project.py

@@ -1,27 +1,13 @@
 from django.core.management.base import BaseCommand
 
-from clickhouse_search.search import delete_clickhouse_family
-from seqr.models import Project, Family, VariantTag, VariantTagType, Sample
+from seqr.models import Project, Family, VariantTag, VariantTagType
 from seqr.utils.search.utils import backend_specific_call
+from seqr.utils.search.add_data_utils import trigger_delete_families_search
 
 import logging
 logger = logging.getLogger(__name__)
 
 
-def _disable_search(families, from_project):
-    search_samples = Sample.objects.filter(is_active=True, individual__family__in=families)
-    if search_samples:
-        updated_families = search_samples.values_list("individual__family__family_id", flat=True).distinct()
-        updated_family_dataset_types = list(search_samples.values_list('individual__family__guid', 'dataset_type', 'sample_type').distinct())
-        family_summary = ", ".join(sorted(updated_families))
-        num_updated = search_samples.update(is_active=False)
-        logger.info(
-            f'Disabled search for {num_updated} samples in the following {len(updated_families)} families: {family_summary}'
-        )
-        for update in updated_family_dataset_types:
-            logger.info(delete_clickhouse_family(from_project, *update))
-
-
 class Command(BaseCommand):
     def add_arguments(self, parser):
         parser.add_argument('--from-project', required=True)
@@ -49,7 +35,7 @@ def handle(self, *args, **options):
             ]
             logger.info(f'Skipping {num_found - len(families)} families with analysis groups in the project: {", ".join(group_families)}')
 
-        backend_specific_call(lambda *args: None, _disable_search)(families, from_project)
+        backend_specific_call(lambda *args: None, trigger_delete_families_search)(from_project, list(families.values_list('guid', flat=True)))
 
         for variant_tag_type in VariantTagType.objects.filter(project=from_project):
             variant_tags = VariantTag.objects.filter(saved_variants__family__in=families, variant_tag_type=variant_tag_type)

seqr/management/tests/transfer_families_to_different_project_tests.py

@@ -1,4 +1,5 @@
 from django.core.management import call_command
+import responses
 
 from seqr.models import Family, VariantTagType, VariantTag, Sample
 from seqr.views.utils.test_utils import AuthenticationTestCase, AnvilAuthenticationTestCase
@@ -9,7 +10,10 @@ class TransferFamiliesTest(object):
     DEACTIVATE_SEARCH = True
     LOGS = []
 
+    @responses.activate
     def test_command(self):
+        responses.add(responses.POST, 'http://pipeline-runner:6000/delete_families_enqueue', status=200)
+
         call_command(
             'transfer_families_to_different_project', '--from-project=R0001_1kg', '--to-project=R0003_test', '2', '4', '5', '12',
         )
@@ -59,7 +63,5 @@ class TransferFamiliesClickhouseTest(TransferFamiliesTest, AnvilAuthenticationTe
     ES_HOSTNAME = ''
     LOGS = [
         ('Disabled search for 7 samples in the following 1 families: 2', None),
-        ('Clickhouse does not support deleting individual families from project. Manually delete MITO data for F000002_2 in project R0001_1kg', None),
-        ('Clickhouse does not support deleting individual families from project. Manually delete SNV_INDEL data for F000002_2 in project R0001_1kg', None),
-        ('Clickhouse does not support deleting individual families from project. Manually delete GCNV data for F000002_2 in project R0001_1kg', None),
+        ('Triggered Delete Families', {'detail':  {'project_guid': 'R0001_1kg', 'family_guids': ['F000002_2', 'F000004_4']}}),
     ]

seqr/utils/search/add_data_utils.py

@@ -13,6 +13,7 @@
 from seqr.utils.middleware import ErrorsWarningsException
 from seqr.views.utils.airtable_utils import AirtableSession, ANVIL_REQUEST_TRACKING_TABLE
 from seqr.views.utils.export_utils import write_multiple_files
+from seqr.views.utils.json_utils import _to_title_case
 from seqr.views.utils.pedigree_info_utils import JsonConstants
 from settings import SEQR_SLACK_DATA_ALERTS_NOTIFICATION_CHANNEL, BASE_URL, ANVIL_UI_URL, PIPELINE_RUNNER_SERVER, \
     SEQR_SLACK_ANVIL_DATA_LOADING_CHANNEL, SEQR_SLACK_LOADING_NOTIFICATION_CHANNEL, LOADING_DATASETS_DIR
@@ -65,6 +66,22 @@ def update_airtable_loading_tracking_status(project, status, additional_update=N
         update={'Status': status, **(additional_update or {})},
     )
 
+def trigger_delete_families_search(project, family_guids, user=None):
+    search_samples = Sample.objects.filter(is_active=True, individual__family__guid__in=family_guids)
+    info = []
+    if search_samples:
+        updated_families = search_samples.values_list("individual__family__family_id", flat=True).distinct()
+        family_summary = ", ".join(sorted(updated_families))
+        num_updated = search_samples.update(is_active=False)
+        message = f'Disabled search for {num_updated} samples in the following {len(updated_families)} families: {family_summary}'
+        info.append(message)
+        logger.info(message, user)
+
+    variables = {'project_guid': project.guid, 'family_guids': family_guids}
+    _enqueue_pipeline_request('delete_families', variables, user)
+    info.append('Triggered delete family data')
+    return info
+
 def trigger_data_loading(projects: list[Project], individual_ids: list[int], sample_type: str, dataset_type: str,
                          genome_version: str, data_path: str, user: User, raise_error: bool = False, skip_expect_tdr_metrics: bool = True,
                          skip_check_sex_and_relatedness: bool = True, vcf_sample_id_map=None,
@@ -85,34 +102,40 @@ def trigger_data_loading(projects: list[Project], individual_ids: list[int], sam
     _upload_data_loading_files(individual_ids, vcf_sample_id_map or {}, user, file_path, raise_error)
     _write_gene_id_file(user)
 
-    response = requests.post(f'{PIPELINE_RUNNER_SERVER}/loading_pipeline_enqueue', json=variables, timeout=60)
-    success = True
+    error = _enqueue_pipeline_request('loading_pipeline', variables, user, raise_error)
+    if error:
+        safe_post_to_slack(
+            SEQR_SLACK_LOADING_NOTIFICATION_CHANNEL,
+            f'{error_message}: {error}\nLoading pipeline should be triggered with:\n```{json.dumps(variables, indent=4)}```',
+        )
+
+    success = not error
+    if success_message and (success or success_slack_channel != SEQR_SLACK_LOADING_NOTIFICATION_CHANNEL):
+        safe_post_to_slack(success_slack_channel, '\n\n'.join([
+            success_message,
+            f'Pedigree files have been uploaded to {file_path}',
+            f'Loading pipeline is triggered with:\n```{json.dumps(variables, indent=4)}```',
+        ]))
+
+    return success
+
+
+def _enqueue_pipeline_request(name: str, variables: dict, user: User, raise_error: bool = True):
+    response = requests.post(f'{PIPELINE_RUNNER_SERVER}/{name}_enqueue', json=variables, timeout=60)
+    error = None
     try:
         response.raise_for_status()
-        logger.info('Triggered loading pipeline', user, detail=variables)
+        logger.info(f'Triggered {_to_title_case(name)}', user, detail=variables)
     except requests.HTTPError as e:
-        success = False
         error = str(e)
         if response.status_code == 409:
             error = 'Loading pipeline is already running. Wait for it to complete and resubmit'
             e = ErrorsWarningsException([error])
         if raise_error:
             raise e
         else:
-            logger.warning(f'Error triggering loading pipeline: {error}', user, detail=variables)
-            safe_post_to_slack(
-                SEQR_SLACK_LOADING_NOTIFICATION_CHANNEL,
-                f'{error_message}: {error}\nLoading pipeline should be triggered with:\n```{json.dumps(variables, indent=4)}```',
-            )
-
-    if success_message and (success or success_slack_channel != SEQR_SLACK_LOADING_NOTIFICATION_CHANNEL):
-        safe_post_to_slack(success_slack_channel, '\n\n'.join([
-            success_message,
-            f'Pedigree files have been uploaded to {file_path}',
-            f'Loading pipeline is triggered with:\n```{json.dumps(variables, indent=4)}```',
-        ]))
-
-    return success
+            logger.warning(f'Error Triggering {_to_title_case(name)}: {error}', user, detail=variables)
+    return error
 
 
 def _loading_dataset_type(sample_type: str, dataset_type: str):

seqr/views/apis/anvil_workspace_api_tests.py

@@ -910,7 +910,7 @@ def _test_mv_file_and_triggering_loading_exception(self, url, workspace, sample_
             self.manager_user, detail={f'{project.guid}_pedigree': sample_data})
         self.mock_api_logger.error.assert_not_called()
         self.mock_add_data_utils_logger.warning.assert_called_with(
-            'Error triggering loading pipeline: Loading pipeline is already running. Wait for it to complete and resubmit',
+            'Error Triggering Loading Pipeline: Loading pipeline is already running. Wait for it to complete and resubmit',
             self.manager_user, detail=variables,
         )
         self.mock_airtable_logger.error.assert_called_with(

seqr/views/apis/data_manager_api.py

@@ -13,9 +13,9 @@
 from django.views.decorators.csrf import csrf_exempt
 from requests.exceptions import ConnectionError as RequestConnectionError
 
-from clickhouse_search.search import delete_clickhouse_project, delete_clickhouse_family
+from clickhouse_search.search import delete_clickhouse_project
 from seqr.utils.communication_utils import send_project_notification
-from seqr.utils.search.add_data_utils import trigger_data_loading, get_loading_samples_validator
+from seqr.utils.search.add_data_utils import trigger_data_loading, get_loading_samples_validator, trigger_delete_families_search
 from seqr.utils.search.elasticsearch.es_utils import get_elasticsearch_status, delete_es_index
 from seqr.utils.search.utils import clickhouse_only, es_only, InvalidSearchException
 from seqr.utils.file_utils import file_iter, does_file_exist
@@ -424,9 +424,17 @@ def fetch_missing_vcf_samples(missing_vcf_samples):
 def trigger_delete_project(request):
     request_json = json.loads(request.body)
     project_guid = request_json.pop('project')
+    dataset_type = request_json.get('datasetType')
     project = Project.objects.get(guid=project_guid)
-    samples = Sample.objects.filter(individual__family__project=project)
-    return _trigger_data_update(delete_clickhouse_project, request.user, samples, request_json, project)
+    samples = Sample.objects.filter(individual__family__project=project, dataset_type=dataset_type, is_active=True)
+    sample_types = list(
+        samples.values_list('sample_type', flat=True).distinct()
+    ) if dataset_type == Sample.DATASET_TYPE_SV_CALLS else [None]
+    updated = Sample.bulk_update(user=request.user, update_json={'is_active': False}, queryset=samples)
+    info = [f'Deactivated search for {len(updated)} individuals']
+    for sample_type in sample_types:
+        info.append(delete_clickhouse_project(project, dataset_type=dataset_type, sample_type=sample_type))
+    return create_json_response({'info': info})
 
 
 @data_manager_required
@@ -436,19 +444,7 @@ def trigger_delete_family(request):
     family_guid = request_json.pop('family')
     project = Project.objects.get(family__guid=family_guid)
     samples = Sample.objects.filter(individual__family__guid=family_guid)
-    return _trigger_data_update(delete_clickhouse_family, request.user, samples, request_json, project, family_guid)
-
-
-def _trigger_data_update(clickhouse_func, user, samples, request_json, project, *args):
-    dataset_type = request_json.get('datasetType')
-    samples = samples.filter(dataset_type=dataset_type, is_active=True)
-    sample_types = list(
-        samples.values_list('sample_type', flat=True).distinct()
-    ) if dataset_type == Sample.DATASET_TYPE_SV_CALLS else [None]
-    updated = Sample.bulk_update(user=user, update_json={'is_active': False}, queryset=samples)
-    info = [f'Deactivated search for {len(updated)} individuals']
-    for sample_type in sample_types:
-        info.append(clickhouse_func(project, *args, dataset_type=dataset_type, sample_type=sample_type))
+    info = trigger_delete_families_search(project, [family_guid], request.user)
     return create_json_response({'info': info})
 
 

seqr/views/apis/data_manager_api_tests.py

@@ -458,7 +458,8 @@
     'ABC123', 'NA19675_1', 'NA19678', 'NA19679', 'HG00731', 'HG00732', 'HG00733', 'NA20874', 'NA21234', 'NA21987',
 ]
 
-PIPELINE_RUNNER_URL = 'http://pipeline-runner:6000/loading_pipeline_enqueue'
+PIPELINE_RUNNER_HOST = 'http://pipeline-runner:6000'
+PIPELINE_RUNNER_URL = f'{PIPELINE_RUNNER_HOST}/loading_pipeline_enqueue'
 
 
 @mock.patch('seqr.views.apis.data_manager_api.LOADING_DATASETS_DIR', '/local_datasets')
@@ -1593,11 +1594,13 @@ def test_trigger_delete_project(self):
 
     @responses.activate
     def test_trigger_delete_family(self):
+        responses.add(responses.POST, f'{PIPELINE_RUNNER_HOST}/delete_families_enqueue', status=200)
+
         url = reverse(trigger_delete_family)
         self.check_data_manager_login(url)
 
         Project.objects.filter(guid=PROJECT_GUID).update(genome_version='38')
-        response = self.client.post(url, content_type='application/json', data=json.dumps({'family': 'F000002_2', 'datasetType': 'SV'}))
+        response = self.client.post(url, content_type='application/json', data=json.dumps({'family': 'F000002_2'}))
         self._assert_expected_delete_family(response)
 
     def _assert_expected_delete_project(self, response):
@@ -1694,7 +1697,7 @@ def _has_expected_ped_files(self, mock_open, mock_gzip_open, mock_mkdir, dataset
 
     def _assert_success_notification(self, variables):
         self.maxDiff = None
-        self.assert_json_logs(self.pm_user, [('Triggered loading pipeline', {'detail': variables})])
+        self.assert_json_logs(self.pm_user, [('Triggered Loading Pipeline', {'detail': variables})])
 
     def _trigger_error(self, url, body, variables, mock_open, mock_gzip_open, mock_mkdir):
         super()._trigger_error(url, body, variables, mock_open, mock_gzip_open, mock_mkdir)
@@ -1915,7 +1918,7 @@ def _assert_trigger_error(self, response, body, variables, response_body):
             'callset_path': variables['callset_path'].replace('callset.vcf', 'sv_callset.vcf'),
         }
         self.assert_json_logs(self.data_manager_user, [
-            (f'Error triggering loading pipeline: {error}', {'severity': 'WARNING', 'detail': variables}),
+            (f'Error Triggering Loading Pipeline: {error}', {'severity': 'WARNING', 'detail': variables}),
         ], offset=6)
 
         error_message = f"""ERROR triggering internal WES SV loading: {error}
@@ -2082,14 +2085,19 @@ def _assert_expected_delete_family(self, response):
         self.assertEqual(response.status_code, 200)
         self.assertDictEqual(response.json(), {
             'info': [
-                'Deactivated search for 3 individuals',
-                'Clickhouse does not support deleting individual families from project. Manually delete GCNV data for F000002_2 in project R0001_1kg',
+                'Disabled search for 7 samples in the following 1 families: 2',
+                'Triggered delete family data',
             ],
         })
 
         family_samples = Sample.objects.filter(individual__family_id=2, is_active=True)
-        self.assertEqual(family_samples.filter(dataset_type='SV').count(), 0)
-        self.assertEqual(family_samples.count(),4)
+        self.assertEqual(family_samples.count(),0)
+
+        self.assertEqual(len(responses.calls), 1)
+        self.assertDictEqual(json.loads(responses.calls[-1].request.body), {
+            'project_guid': 'R0001_1kg',
+            'family_guids': ['F000002_2'],
+        })
 
     def _assert_expected_airtable_errors(self, url):
         responses.replace(

ui/pages/DataManagement/components/TriggerSearchDataUpdatePages.jsx

@@ -43,7 +43,6 @@ const FAMILY_FIELDS = [
     placeholder: 'Search for a family',
     validate: validators.required,
   },
-  DATASET_TYPE_FIELD,
 ]
 
 const TriggerSearchDataUpdateForm = ({ entity, fields }) => (