Merge pull request #4626 from fedspendingtransparency/fix/dev-14824-handle-datetime-format-on-load

[DEV-14824] Downmerge into QAT

Author: sethstoudenmier

usaspending_api/common/helpers/spark_helpers.py

@@ -708,7 +708,7 @@ def clean_postgres_sql_for_spark_sql(
     if identifier_replacements:
         for old, new in identifier_replacements.items():
             matches = re.finditer(
-                rf"(\s+|^|\(|'){old}(\s+|$|\(|')",
+                rf"(\s+|^|\(){old}(\s+|$|\()",
                 spark_sql,
                 flags=re.IGNORECASE | re.MULTILINE,
             )

usaspending_api/database_scripts/etl/award_delta_view.sql

@@ -20,7 +20,7 @@ SELECT
   "award_amount",
   "total_subsidy_cost",
   "total_loan_value",
-  CAST(TO_CHAR(update_date, 'yyyy-MM-dd HH24:MI:SS') AS TIMESTAMP) AS update_date,
+  "update_date",
 
   "recipient_name",
   "recipient_unique_id",
@@ -34,7 +34,7 @@ SELECT
 
   "action_date",
   "fiscal_year",
-  CAST(TO_CHAR(last_modified_date, 'yyyy-MM-dd HH24:MI:SS') AS TIMESTAMP) AS last_modified_date,
+  "last_modified_date",
   "period_of_performance_start_date",
   "period_of_performance_current_end_date",
   "date_signed",

usaspending_api/database_scripts/etl/subaward_es_view.sql

@@ -100,7 +100,7 @@ SELECT
 	s.prime_award_group,
 	s.prime_award_type,
 	s.latest_transaction_id,
-	CAST(TO_CHAR(s.last_modified_date, 'yyyy-MM-dd HH24:MI:SS') AS TIMESTAMP) AS last_modified_date,
+	s.last_modified_date,
 	s.awarding_toptier_agency_name,
 	s.awarding_toptier_agency_abbreviation,
 	s.funding_toptier_agency_name,

usaspending_api/database_scripts/etl/transaction_delta_view.sql

@@ -23,14 +23,14 @@ SELECT
   END AS display_award_id,
   "action_date",
   "fiscal_action_date",
-  CAST(TO_CHAR(last_modified_date, 'yyyy-MM-dd HH24:MI:SS') AS TIMESTAMP) AS last_modified_date,
+  "last_modified_date",
   "fiscal_year",
   "award_certified_date",
   "award_fiscal_year",
   "award_date_signed",
-  CAST(TO_CHAR(update_date, 'yyyy-MM-dd HH24:MI:SS') AS TIMESTAMP) AS update_date,
+  "update_date",
   "award_update_date",
-  CAST(TO_CHAR(etl_update_date, 'yyyy-MM-dd HH24:MI:SS') AS TIMESTAMP) AS etl_update_date,
+  "etl_update_date",
   "period_of_performance_start_date",
   "period_of_performance_current_end_date",
   "ordering_period_end_date",

usaspending_api/etl/elasticsearch_loader_helpers/controller_for_spark.py

@@ -77,22 +77,13 @@ def ensure_view_exists(
             f"DROP VIEW IF EXISTS {sql_view_name};", ""
         )
 
-        # Replacements to supports converting timestamp to a formatted string in both Postgres and Spark
-        datetime_replacements = {
-            "TO_CHAR": "DATE_FORMAT",
-            "yyyy-MM-dd HH24:MI:SS": "yyyy-MM-dd HH:mm:ss",  # different formats to support 24-hour time
-        }
-
         identifier_replacements = {}
         if self.config["load_type"] == "transaction":
             identifier_replacements["transaction_search"] = "rpt.transaction_search"
-            identifier_replacements.update(datetime_replacements)
         elif self.config["load_type"] == "award":
             identifier_replacements["award_search"] = "rpt.award_search"
-            identifier_replacements.update(datetime_replacements)
         elif self.config["load_type"] == "subaward":
             identifier_replacements["toptier_agency"] = "global_temp.toptier_agency"
-            identifier_replacements.update(datetime_replacements)
         elif self.config["load_type"] == "recipient":
             identifier_replacements = None
         else:

usaspending_api/etl/elasticsearch_loader_helpers/transform_data.py

@@ -10,6 +10,7 @@
     convert_json_array_to_list_of_str,
     convert_json_data_to_dict,
     format_log,
+    format_timestamp_as_string,
 )
 
 logger = logging.getLogger("script")
@@ -19,7 +20,9 @@ def transform_award_data(worker: TaskSpec, records: List[dict]) -> List[dict]:
     replace_fields = {
         "spending_by_defc": convert_json_data_to_dict,
         "federal_accounts": convert_json_array_to_list_of_str,
+        "last_modified_date": format_timestamp_as_string,
         "program_activities": convert_json_data_to_dict,
+        "update_date": format_timestamp_as_string,
     }
     # TODO: Move some of the 1:1 agg_keys that match a field already on Elasticsearch
     insert_fields = {
@@ -64,8 +67,11 @@ def transform_award_data(worker: TaskSpec, records: List[dict]) -> List[dict]:
 
 def transform_transaction_data(worker: TaskSpec, records: List[dict]) -> List[dict]:
     replace_fields = {
+        "etl_update_date": format_timestamp_as_string,
         "federal_accounts": convert_json_array_to_list_of_str,
+        "last_modified_date": format_timestamp_as_string,
         "program_activities": convert_json_data_to_dict,
+        "update_date": format_timestamp_as_string,
     }
     # TODO: Move some of the 1:1 agg_keys that match a field already on Elasticsearch
     insert_fields = {
@@ -108,6 +114,7 @@ def transform_transaction_data(worker: TaskSpec, records: List[dict]) -> List[di
 
 def transform_subaward_data(worker: TaskSpec, records: List[dict]) -> List[dict]:
     replace_fields = {
+        "last_modified_date": format_timestamp_as_string,
         "program_activities": convert_json_data_to_dict,
     }
     insert_fields = {

usaspending_api/etl/elasticsearch_loader_helpers/utilities.py

@@ -2,6 +2,7 @@
 import logging
 import re
 from dataclasses import dataclass
+from datetime import datetime
 from pathlib import Path
 from random import choice
 from typing import Any, Generator, List, Optional, Union
@@ -136,6 +137,10 @@ def convert_json_array_to_list_of_str(json_data: Union[list, str]) -> Optional[L
     return result
 
 
+def format_timestamp_as_string(value_to_format: datetime, datetime_format: str = "%Y-%m-%d %H:%M:%S") -> str:
+    return datetime.strftime(value_to_format, datetime_format) if value_to_format else value_to_format
+
+
 def execute_sql_statement(cmd: str | Composed, results: bool = False, verbose: bool = False) -> Optional[List[dict]]:
     """Simple function to execute SQL using a single-use psycopg connection"""
     rows = None

usaspending_api/etl/management/commands/elasticsearch_indexer.py

@@ -241,10 +241,6 @@ def parse_cli_args(options: dict, es_client: Elasticsearch) -> dict:  # noqa: PL
         logger.error(f"--start-datetime is too early. Set no earlier than {config['initial_datetime']}")
         raise SystemExit(1)
 
-    # Format to include timezone, but remove milliseconds to allow proper comparison with datetime values
-    # that are captured on the documents
-    config["starting_date"] = datetime.strftime(config["starting_date"], "%Y-%m-%d %H:%M:%S%z")
-
     return config