[DEV-12639] fix duplicate awards

Author: loreleitrimberger

usaspending_api/broker/helpers/last_load_date.py

@@ -1,7 +1,6 @@
 import logging
 
 from datetime import timedelta
-from itertools import islice
 
 from usaspending_api.broker import lookups
 from usaspending_api.broker.models import ExternalDataLoadDate
@@ -99,34 +98,3 @@ def update_last_load_date(key, last_load_date):
         external_data_type_id=lookups.EXTERNAL_DATA_TYPE_DICT[key],
         defaults={"last_load_date": cast_datetime_to_utc(last_load_date)},
     )
-
-def get_second_to_last_load_data(key, lookback_minutes=None, default=None, format_func: callable = (lambda _: _)):
-    """
-    Retrieve the second_to_last_load_date from the USAspending database.
-
-    Valid keys are dictated by the keys in EXTERNAL_DATA_TYPE_DICT.
-
-    lookback_minutes is used to provide some protection against gaps caused by
-    long transactions or race conditions.  It will be subtracted from
-    second_to_last_load_data.  NOTE:  It will not be subtracted from the default in the
-    case where no second_to_last_load_data is found.
-
-    default will be returned if no second_to_last_load_data is found in the database.
-    """
-    external_data_type_id = lookups.EXTERNAL_DATA_TYPE_DICT[key]
-    load_date = (
-        ExternalDataLoadDate.objects.filter(external_data_type_id=external_data_type_id)
-        .values_list("last_load_date", flat=True)
-    )
-    if load_date is None:
-        logger.warning(format_func(f"No record of a previous run for `{key}` was found!"))
-        return default
-    second_to_last_load_data = next(islice(load_date.iterator(), 1, 2), None)
-    if second_to_last_load_data is None:
-        logger.warning(format_func(f"No record of a previous run for `{key}` was found!"))
-        return default
-    else:
-        logger.info(format_func(f"Value for previous `{key}` ETL: {second_to_last_load_data}"))
-    if lookback_minutes is not None:
-        second_to_last_load_data -= timedelta(minutes=lookback_minutes)
-    return second_to_last_load_data

usaspending_api/download/tests/integration/test_account_download_dataframe_builder.py

@@ -250,4 +250,3 @@ def test_account_balances(mock_get_submission_ids_for_periods, spark, account_do
     assert ta_builder.account_balances.count() == 2
     fa_builder = FederalAccountDownloadDataFrameBuilder(spark, account_download_filter)
     assert fa_builder.account_balances.count() == 2
-

usaspending_api/etl/elasticsearch_loader_helpers/delete_data.py

@@ -1,4 +1,5 @@
 import logging
+import re
 from datetime import datetime
 from time import perf_counter
 from typing import Dict, List, Optional, Union
@@ -14,7 +15,6 @@
 from usaspending_api.broker.helpers.last_load_date import (
     get_last_load_date,
     get_latest_load_date,
-    get_second_to_last_load_data
 )
 from usaspending_api.common.helpers.s3_helpers import (
     access_s3_object,
@@ -316,15 +316,11 @@ def delete_awards(
 
     Returns: Number of ES docs deleted in the index
     """
-    es_delete_window_start = get_second_to_last_load_data(
-        "es_deletes", format_func=(lambda log_msg: format_log(log_msg, action="Delete"))
-    )
-
     delete_window_start = get_last_load_date(
         "es_deletes", format_func=(lambda log_msg: format_log(log_msg, action="Delete"))
     )
     deleted_tx_keys = _gather_deleted_transaction_keys(
-        config, es_delete_window_start, fabs_external_data_load_date_key, fpds_external_data_load_date_key
+        config, delete_window_start, fabs_external_data_load_date_key, fpds_external_data_load_date_key
     )
     awards_to_delete = []
 
@@ -393,14 +389,12 @@ def delete_transactions(
 
     tx_keys_to_delete = []
 
-    es_delete_window_start = get_second_to_last_load_data("es_deletes", format_func=(lambda log_msg: format_log(log_msg, action="Delete")))
-
     delete_window_start = get_last_load_date(
         "es_deletes", format_func=(lambda log_msg: format_log(log_msg, action="Delete"))
     )
 
     deleted_tx_keys = _gather_deleted_transaction_keys(
-        config, es_delete_window_start, fabs_external_data_load_date_key, fpds_external_data_load_date_key
+        config, delete_window_start, fabs_external_data_load_date_key, fpds_external_data_load_date_key
     )
     tx_keys_to_delete.extend(deleted_tx_keys)
 
@@ -471,6 +465,12 @@ def _gather_deleted_transaction_keys(
             and not x.key.startswith("staging")
             and delete_window_start <= x.last_modified <= end_date
         )
+        # matches the pattern YYYY-mm-DD_FABSdeletions_{TIME_IN_SECONDS}.csv
+        or (
+            re.search(r"^\d{4}-(0[1-9]|1[0-2])-(0?[1-9]|[12][0-9]|3[01])_FABSdeletions_.*.csv", x.key)
+            and not x.key.startswith("staging")
+            and delete_window_start <= x.last_modified
+        )
     ]
 
     logger.info(format_log(f"Found {len(filtered_csv_list)} CSV files in the date range", action="Delete"))

usaspending_api/etl/management/commands/load_transactions_in_delta.py

@@ -86,8 +86,7 @@ class Command(BaseCommand):
         FROM aidlu_fabs AS aidlu LEFT JOIN raw.published_fabs AS pfabs ON (
             aidlu.transaction_unique_id = ucase(pfabs.afa_generated_unique)
         )
-        WHERE pfabs.afa_generated_unique IS NULL AND NOT EXISTS (SELECT 1 FROM aidlu_fabs
-        WHERE aidlu.transaction_unique_id = ucase(pfabs.afa_generated_unique) AND pfabs.correction_delete_indicatr = null)
+        WHERE pfabs.afa_generated_unique IS NULL
     """
 
     def add_arguments(self, parser):