Allow bq-ingester to either read parsed_files as being a JSON column parsed into a map by the bq client, or as JSON column containing a JSON-encoded string, returned from the client as an unparsed string. (#269)

Author: theferrit32

clinvar_ingest/cloud/bigquery/processing_history.py

@@ -295,7 +295,7 @@ def write_started(  # noqa: PLR0913
         if row.c != 0:
             if error_if_exists:
                 raise RuntimeError(
-                    f"Expected 1 row to exist for the finished event, but found {row.c}. "
+                    f"Expected 0 row to exist for the finished event, but found {row.c}. "
                     f"file_type={file_type}, release_date={release_date}, "
                     f"release_tag={release_tag}, bucket_dir={bucket_dir}"
                 )

misc/bin/bq_ingester.py

@@ -8,6 +8,7 @@
 import json
 import logging
 import sys
+
 from google.cloud import bigquery
 from google.cloud.storage import Client as GCSClient
 
@@ -72,8 +73,10 @@ def _get_bq_client() -> bigquery.Client:
     processing_history_table=processing_history_table,
     client=_get_bq_client(),
 )
-processing_history_needing_bq_ingest = processing_history.processed_entries_ready_for_bq_ingest(
-    processing_history_view, client=_get_bq_client()
+processing_history_needing_bq_ingest = (
+    processing_history.processed_entries_ready_for_bq_ingest(
+        processing_history_view, client=_get_bq_client()
+    )
 )
 msg = f"Found {processing_history_needing_bq_ingest.total_rows} VCV/RCV datasets to ingest."
 _logger.info(msg)
@@ -83,6 +86,13 @@ def _get_bq_client() -> bigquery.Client:
 
 send_slack_message(msg)
 
+
+def json_parse_if_string(value):
+    if isinstance(value, str):
+        return json.loads(value)
+    return value
+
+
 # update processing_history to set the bq ingest as having started
 rows_to_ingest = []
 for row in processing_history_needing_bq_ingest:
@@ -100,7 +110,7 @@ def _get_bq_client() -> bigquery.Client:
         client=_get_bq_client(),
         bucket_dir=vcv_bucket_dir,
         xml_release_date=str(vcv_xml_release_date),
-        error_if_exists=True,
+        error_if_exists=False,
     )
     _logger.info(
         f"Created bq_ingest_processing record with version {vcv_pipeline_version} and "
@@ -119,7 +129,7 @@ def _get_bq_client() -> bigquery.Client:
     vcv_release_date = row.get("vcv_release_date", None)
     vcv_xml_release_date = row.get("vcv_xml_release_date", None)
     vcv_bucket_dir = row.get("vcv_bucket_dir", None)
-    vcv_parsed_files = json.loads(row.get("vcv_parsed_files", None))
+    vcv_parsed_files = json_parse_if_string(row.get("vcv_parsed_files", None))
 
     rcv_file_type = row.get("rcv_file_type", None)
     rcv_pipeline_version = row.get("rcv_pipeline_version", None)
@@ -129,7 +139,7 @@ def _get_bq_client() -> bigquery.Client:
     rcv_release_date = row.get("rcv_release_date", None)
     rcv_xml_release_date = row.get("rcv_xml_release_date", None)
     rcv_bucket_dir = row.get("rcv_bucket_dir", None)
-    rcv_parsed_files = json.loads(row.get("rcv_parsed_files", None))
+    rcv_parsed_files = json_parse_if_string(row.get("rcv_parsed_files", None))
 
     # We will use the VCV file's release date as the "release date" for both
     # And the VCV pipeline version as the "pipeline version"
@@ -173,9 +183,7 @@ def _get_bq_client() -> bigquery.Client:
             vcv_create_tables_request
         )
         vcv_ext_resp_json = json.dumps(
-            walk_and_replace(
-                vcv_create_tables_response, processing_history._dump_fn
-            )
+            walk_and_replace(vcv_create_tables_response, processing_history._dump_fn)
         )
         _logger.info(f"VCV Create External Tables response: {vcv_ext_resp_json}")
 
@@ -191,9 +199,7 @@ def _get_bq_client() -> bigquery.Client:
             rcv_create_tables_request
         )
         rcv_ext_resp_json = json.dumps(
-            walk_and_replace(
-                rcv_create_tables_response, processing_history._dump_fn
-            )
+            walk_and_replace(rcv_create_tables_response, processing_history._dump_fn)
         )
         _logger.info(f"RCV Create External Tables response: {rcv_ext_resp_json}")