COLIN Extract Dynamic Schema Name Changes

data-tool/Makefile

@@ -89,15 +89,6 @@ run-prefect-server: ## Start Prefect server
 run-prefect-reset-db: ## clears all data and reapplies the schema.  Handy for clearing flow run history.
 	. $(PREFECT_VENV_DIR)/bin/activate && prefect server database reset -y
 
-run-corps-migration: ## Run migration flow
-	@echo "Current directory: $(CURRENT_ABS_DIR)"
-	@echo "SQLAlchemy path: $(CURRENT_ABS_DIR)/$(SQLALCHEMY_VENV_DIR)/lib/$(PYTHON_VERSION)/site-packages"
-	. $(VENV_DIR)/bin/activate && \
-	export PYTHONPATH="$(CURRENT_ABS_DIR):$$PYTHONPATH" && \
-	export SQLALCHEMY_PATH="$(CURRENT_ABS_DIR)/$(SQLALCHEMY_VENV_DIR)/lib/$(PYTHON_VERSION)/site-packages" && \
-	FLASK_ENV=development && \
-	python flows/migrate_corps_flow.py
-
 run-corps-delete: ## Run delete flow
 	. $(VENV_DIR)/bin/activate && \
 	python flows/batch_delete_flow.py

data-tool/flows/common/colin_queries.py

@@ -69,20 +69,20 @@ def get_updated_identifiers(timestamp: str, corp_list: str, chunk_size: int, sco
     """
     return query
 
-def get_identifiers_per_batch(mig_batch_id: int) -> str:
+def get_identifiers_per_batch(mig_batch_id: int, target_schema: str) -> str:
     return f"""
     SELECT string_agg(pg_catalog.quote_literal(trim(CAST(mcb.corp_num AS text))), ',') AS corp_list
-    FROM mig_corp_batch mcb
+    FROM {target_schema}.mig_corp_batch mcb
     WHERE mcb.mig_batch_id IN ({mig_batch_id})
     """
 
-def unfreeze_identifiers() -> str:
+def unfreeze_identifiers(target_schema: str) -> str:
     return f"""
-    UPDATE corporation AS c
-    SET corp_frozen_type_cd = NULL
-    FROM mig_group AS mg
-            JOIN mig_batch AS mb ON mb.mig_group_id = mg.id
-            JOIN mig_corp_batch AS mcb ON mcb.mig_batch_id = mb.id
+    UPDATE {target_schema}.corporation AS c
+    SET {target_schema}.corp_frozen_type_cd = NULL
+    FROM {target_schema}.mig_group AS mg
+            JOIN {target_schema}.mig_batch AS mb ON mb.mig_group_id = mg.id
+            JOIN {target_schema}.mig_corp_batch AS mcb ON mcb.mig_batch_id = mb.id
     WHERE c.corp_num = mcb.corp_num
     -- cprd
     and mg.name in ('group_0', 'group_1', 'group_3', 'group_4','gcp_migration_group_test','misc_group')

data-tool/flows/common/query_utils.py

@@ -53,14 +53,14 @@ def corpnum_to_oracle_ids(target_ids: str | bytes | tuple | list | None) -> List
         return None
     return ",".join("'" + x.replace("'", "''") + "'" for x in out)
 
-def get_candidates_not_matching_saf_criteria_query(updated_corp_nums: list) -> str:
+def get_candidates_not_matching_saf_criteria_query(updated_corp_nums: list, target_schema: str) -> str:
     in_list = colin_oracle_corp_num_list_format(updated_corp_nums)
     return f"""
-    SELECT corp_num FROM mv_legacy_corps_data
+    SELECT corp_num FROM {target_schema}.mv_legacy_corps_data
     WHERE 1 = 1
     AND corp_num IN {in_list}
     AND corp_num NOT IN (
-    SELECT corp_num FROM mv_legacy_corps_data
+    SELECT corp_num FROM {target_schema}.mv_legacy_corps_data
     WHERE 1 = 1
     AND is_active = true
     AND is_frozen = false
@@ -82,34 +82,34 @@ def get_candidates_not_matching_saf_criteria_query(updated_corp_nums: list) -> s
     )
 """
 
-def get_fallout_corp_nums(criteria: str, updated_corp_nums: list) -> str:
+def get_fallout_corp_nums(criteria: str, updated_corp_nums: list, target_schema: str) -> str:
     key = (criteria or '').strip().upper()
     if key == 'SAF':
-        return get_candidates_not_matching_saf_criteria_query(updated_corp_nums)
+        return get_candidates_not_matching_saf_criteria_query(updated_corp_nums, target_schema)
     raise ValueError(f'unsupported criteria: {criteria}')
 
-def prune_candidates_from_cp(pruning_corps_list: list) -> str:
+def prune_candidates_from_cp(pruning_corps_list: list, target_schema: str) -> str:
     in_list = colin_oracle_corp_num_list_format(pruning_corps_list)
     return f"""
-    DELETE FROM corp_processing
+    DELETE FROM {target_schema}.corp_processing
     WHERE corp_num IN {in_list}
     """
 
-def prune_candidates_from_batch(pruning_corps_list: list) -> str:
+def prune_candidates_from_batch(pruning_corps_list: list, target_schema: str) -> str:
     in_list = colin_oracle_corp_num_list_format(pruning_corps_list)
     return f"""
-    DELETE FROM mig_corp_batch
+    DELETE FROM {target_schema}.mig_corp_batch
     WHERE corp_num IN {in_list}
     """
 
-def prune_candidates_from_account(pruning_corps_list: list) -> str:
+def prune_candidates_from_account(pruning_corps_list: list, target_schema: str) -> str:
     in_list = colin_oracle_corp_num_list_format(pruning_corps_list)
     return f"""
-    DELETE FROM mig_corp_account
+    DELETE FROM {target_schema}.mig_corp_account
     WHERE corp_num IN {in_list}
     """
 
-def get_cutoff_timestamp_query() -> str:
+def get_cutoff_timestamp_query(target_schema: str) -> str:
     return f"""
-    SELECT extracted_at FROM colin_extract_version
+    SELECT extracted_at FROM {target_schema}.colin_extract_version
     """

data-tool/flows/config.py

@@ -110,6 +110,7 @@ class _Config():  # pylint: disable=too-few-public-methods
 
     DATA_LOAD_ENV = os.getenv('DATA_LOAD_ENV', '')
     TARGET_CONNECTION = os.getenv('TARGET_CONNECTION', 'ctst_pg')
+    TARGET_SCHEMA = os.getenv('TARGET_SCHEMA', '')
     CORP_NAME_SUFFIX = os.getenv('CORP_NAME_SUFFIX', '')
     UPDATE_ENTITY = os.getenv('UPDATE_ENTITY', 'False') == 'True'
     AFFILIATE_ENTITY = os.getenv('AFFILIATE_ENTITY', 'False') == 'True'