BFD-4810: Implement Prior Auth Sync Strategy

apps/bfd-db-migrator-ng/migrations/V1_40_0__add_utn_pattern_idx.sql

@@ -0,0 +1,2 @@
+CREATE INDEX ON idr.prior_auth (utn) WHERE utn NOT LIKE '-%';
+CREATE INDEX ON idr.prior_auth_item (utn) WHERE utn NOT LIKE '-%';

apps/bfd-pipeline-idr/load-credentials.sh

@@ -12,6 +12,8 @@ IDR_WAREHOUSE="$(aws ssm get-parameter --name /bfd/${BFD_ENV}/idr-pipeline/sensi
 export IDR_WAREHOUSE
 IDR_DATABASE="$(aws ssm get-parameter --name /bfd/${BFD_ENV}/idr-pipeline/sensitive/idr_database --with-decryption --query "Parameter.Value" --output text)"
 export IDR_DATABASE
+IDR_EDP_DATABASE="$(aws ssm get-parameter --name /bfd/${BFD_ENV}/idr-pipeline/sensitive/idr_edp_database --with-decryption --query "Parameter.Value" --output text)"
+export IDR_EDP_DATABASE
 IDR_SCHEMA="$(aws ssm get-parameter --name /bfd/${BFD_ENV}/idr-pipeline/sensitive/idr_schema --with-decryption --query "Parameter.Value" --output text)"
 export IDR_SCHEMA
 

apps/bfd-pipeline-idr/load-synthetic-env.sh

@@ -35,13 +35,10 @@ export IDR_WAREHOUSE
 IDR_DATABASE="$(aws ssm get-parameter --name /bfd/${BFD_ENV}/idr-pipeline/sensitive/synthetic_env_database --with-decryption --query "Parameter.Value" --output text)"
 readonly IDR_DATABASE
 export IDR_DATABASE
-IDR_SCHEMA="$(aws ssm get-parameter --name /bfd/${BFD_ENV}/idr-pipeline/sensitive/synthetic_env_schema --with-decryption --query "Parameter.Value" --output text)"
-readonly IDR_SCHEMA
-export IDR_SCHEMA
 
 args=('--load-type' 'initial' '--source' 'snowflake' '--load-mode' 'synthetic')
 if [[ -n "$1" ]]; then
   args+=('--seed-from' "$1")
 fi
 
-IDR_ENABLE_DATE_PARTITIONS=0 IDR_ENABLE_PRIOR_AUTH=1 uv run idr-pipeline "${args[@]}"
+IDR_ENABLE_DATE_PARTITIONS=0 uv run idr-pipeline "${args[@]}"

apps/bfd-pipeline-idr/mock-idr.sql

@@ -871,7 +871,5 @@ CREATE TABLE cms_edp_view_cvm_prau_prd.prauc (
     mr_count_end_dt DATE,
     att_phy_npi VARCHAR(10) NOT NULL,
     rrb_excl_ind VARCHAR(1),
-    idr_insrt_ts TIMESTAMPTZ,
-    idr_updt_ts TIMESTAMPTZ,
     PRIMARY KEY(mbi_num, utn, current_segment)
 );

apps/bfd-pipeline-idr/run-db.sh

@@ -18,15 +18,14 @@ function do_load() {
   PGPASSWORD="$DB_PASSWORD" psql "host=$DB_ENDPOINT port=5432 dbname=fhirdb user=$DB_USERNAME" -f "$SCRIPT_DIR/mock-idr.sql"
   docker exec -u postgres bfd-idr-db psql fhirdb bfd -c "VACUUM FULL ANALYZE"
   BFD_DB_USERNAME="$DB_USERNAME" \
-    BFD_DB_PASSWORD="$DB_PASSWORD" \
-    BFD_DB_ENDPOINT="$DB_ENDPOINT" \
-    IDR_ENABLE_DATE_PARTITIONS=0 \
-    IDR_ENABLE_PRIOR_AUTH=1 \
-    uv run idr-pipeline \
-    --source postgres \
-    --load-mode synthetic \
-    --load-type initial \
-    --seed-from "${1:-"${SCRIPT_DIR}/../bfd-model-idr/out"}"
+  BFD_DB_PASSWORD="$DB_PASSWORD" \
+  BFD_DB_ENDPOINT="$DB_ENDPOINT" \
+  IDR_ENABLE_DATE_PARTITIONS=0 \
+  uv run idr-pipeline \
+  --source postgres \
+  --load-mode synthetic \
+  --load-type initial \
+  --seed-from "${1:-"${SCRIPT_DIR}/../bfd-model-idr/out"}"
 }
 
 image=postgres:16.6

apps/bfd-pipeline-idr/src/idr_pipeline/constants.py

@@ -1,7 +1,12 @@
 from dateutil.relativedelta import relativedelta
 
 from .load_partition import LoadPartition, LoadPartitionGroup, PartitionType
-from .settings import PARTITION_TYPE
+from .settings import IDR_DATABASE, IDR_EDP_DATABASE, PARTITION_TYPE
+
+
+def _period_delimited(*segments: str) -> str:
+    return ".".join(segment for segment in segments if segment)
+
 
 DEFAULT_MAX_DATE = "9999-12-31"
 DEFAULT_MIN_DATE = "0001-01-01"
@@ -25,7 +30,7 @@
 MCS_CLM_SOURCE = "22000"
 VMS_CLM_SOURCE = "23000"
 
-IDR_PREFIX = "cms_vdm_view_mdcr_prd"
+IDR_PREFIX = _period_delimited(IDR_DATABASE, "cms_vdm_view_mdcr_prd")
 IDR_BENE_HISTORY_TABLE = f"{IDR_PREFIX}.v2_mdcr_bene_hstry"
 IDR_BENE_MBI_TABLE = f"{IDR_PREFIX}.v2_mdcr_bene_mbi_id"
 IDR_BENE_XREF_TABLE = f"{IDR_PREFIX}.v2_mdcr_bene_xref"
@@ -65,7 +70,7 @@
 IDR_CONTRACT_PBP_CONTACT_TABLE = f"{IDR_PREFIX}.v2_mdcr_cntrct_pbp_cntct"
 IDR_CONTRACT_PBP_SEGMENT_TABLE = f"{IDR_PREFIX}.v2_mdcr_cntrct_pbp_sgmt"
 
-IDR_PRIOR_AUTH_PREFIX = "cms_edp_view_cvm_prau_prd"
+IDR_PRIOR_AUTH_PREFIX = _period_delimited(IDR_EDP_DATABASE, "cms_edp_view_cvm_prau_prd")
 IDR_PRIOR_AUTH_TABLE = f"{IDR_PRIOR_AUTH_PREFIX}.prauc"
 
 DEATH_DATE_CUTOFF_YEARS = 4

apps/bfd-pipeline-idr/src/idr_pipeline/extractor.py

@@ -33,9 +33,7 @@
     BATCH_MULTIPLIER,
     ENABLE_DATE_PARTITIONS,
     IDR_ACCOUNT,
-    IDR_DATABASE,
     IDR_PRIVATE_KEY,
-    IDR_SCHEMA,
     IDR_USERNAME,
     IDR_WAREHOUSE,
     MIN_BATCH_COMPLETION_DATE,
@@ -192,6 +190,14 @@ def extract_idr_data(
             {"timestamp": compare_timestamp},
         )
 
+    def extract_full_idr_data(self, source: Source) -> Iterator[list[T]]:
+        start_time = self.cls.model_type().min_transaction_date
+        fetch_query = self.get_query(start_time, source)
+        logger.info("extracting full {}", self.cls.table())
+        return self.extract_many(
+            fetch_query.replace("{MIN_TS}", "%(timestamp)s"), {"timestamp": start_time}
+        )
+
     def _transform(self, batch: list[dict[str, DbType]]) -> list[T]:
         self.transform_timer.start()
         res = self.type_adapter.validate_python(
@@ -318,8 +324,6 @@ def connect() -> SnowflakeConnection:
             private_key=private_key_bytes,
             account=IDR_ACCOUNT,
             warehouse=IDR_WAREHOUSE,
-            database=IDR_DATABASE,
-            schema=IDR_SCHEMA,
         )
 
     @override
@@ -379,8 +383,6 @@ def __init__(self) -> None:
                 "user": IDR_USERNAME,
                 "private_key": private_key_bytes,  # type: ignore
                 "warehouse": IDR_WAREHOUSE,
-                "database": IDR_DATABASE,
-                "schema": IDR_SCHEMA,
             }
         ).create()
         self.conn = SnowflakeExtractor.connect()