Certificate of title import for noi (#1138)

certificate of title import for noi

bin/migrate-oats-data/noi/notice_of_intent_migration.py

@@ -28,6 +28,7 @@
 from .notice_of_intent_submissions.parcels import (
     init_notice_of_intent_parcels,
     clean_parcels,
+    process_notice_of_intent_certificate_of_title,
 )
 
 
@@ -74,5 +75,7 @@ def process_notice_of_intent(batch_size):
 
     init_notice_of_intent_parcels(batch_size)
 
+    process_notice_of_intent_certificate_of_title(batch_size)
+
     # this script must be the last one
     process_notice_of_intent_submission_status_emails()

bin/migrate-oats-data/noi/notice_of_intent_submissions/parcels/__init__.py

@@ -2,3 +2,6 @@
     init_notice_of_intent_parcels,
     clean_parcels,
 )
+from .notice_of_intent_certificate_of_title import (
+    process_notice_of_intent_certificate_of_title,
+)

bin/migrate-oats-data/noi/notice_of_intent_submissions/parcels/notice_of_intent_certificate_of_title.py

@@ -0,0 +1,90 @@
+from common import (
+    setup_and_get_logger,
+    BATCH_UPLOAD_SIZE,
+)
+from db import inject_conn_pool
+from psycopg2.extras import RealDictCursor, execute_batch
+
+etl_name = "process_notice_of_intent_certificate_of_title"
+logger = setup_and_get_logger(etl_name)
+
+
+@inject_conn_pool
+def process_notice_of_intent_certificate_of_title(
+    conn=None, batch_size=BATCH_UPLOAD_SIZE
+):
+    logger.info(f"Start {etl_name}")
+    with conn.cursor(cursor_factory=RealDictCursor) as cursor:
+        with open(
+            "noi/sql/notice_of_intent_submission/parcels/certificate_of_title/notice_of_intent_certificate_of_title_count.sql",
+            "r",
+            encoding="utf-8",
+        ) as sql_file:
+            count_query = sql_file.read()
+            cursor.execute(count_query)
+            count_total = dict(cursor.fetchone())["count"]
+        logger.info(f"Total Notice of Intents data to update: {count_total}")
+
+        failed_inserts = 0
+        successful_updates_count = 0
+        last_document_id = 0
+
+        with open(
+            "noi/sql/notice_of_intent_submission/parcels/certificate_of_title/notice_of_intent_certificate_of_title.sql",
+            "r",
+            encoding="utf-8",
+        ) as sql_file:
+            application_sql = sql_file.read()
+            while True:
+                cursor.execute(
+                    f"{application_sql} AND od.document_id > {last_document_id} ORDER BY od.document_id;"
+                )
+
+                rows = cursor.fetchmany(batch_size)
+
+                if not rows:
+                    break
+                try:
+                    records_to_be_updated_count = len(rows)
+
+                    _update_records(conn, batch_size, cursor, rows)
+
+                    successful_updates_count = (
+                        successful_updates_count + records_to_be_updated_count
+                    )
+                    last_document_id = dict(rows[-1])["document_id"]
+
+                    logger.debug(
+                        f"retrieved/updated items count: {records_to_be_updated_count}; total successfully updated notice of intents so far {successful_updates_count}; last updated alr_application_id: {last_document_id}"
+                    )
+                except Exception as err:
+                    # this is NOT going to be caused by actual data update failure. This code is only executed when the code error appears or connection to DB is lost
+                    logger.exception()
+                    conn.rollback()
+                    failed_inserts = count_total - successful_updates_count
+                    last_document_id = last_document_id + 1
+
+    logger.info(
+        f"Finished {etl_name}: total amount of successful updates {successful_updates_count}, total failed updates {failed_inserts}"
+    )
+
+
+def _update_records(conn, batch_size, cursor, rows):
+    # parsed_data_list = _prepare_oats_data(rows)
+
+    if len(rows) > 0:
+        execute_batch(
+            cursor,
+            _update_query,
+            rows,
+            page_size=batch_size,
+        )
+
+    conn.commit()
+
+
+_update_query = """
+                    UPDATE alcs.notice_of_intent_parcel 
+                            SET certificate_of_title_uuid  = %(document_uuid)s
+                    WHERE oats_subject_property_id = %(subject_property_id)s
+"""

bin/migrate-oats-data/noi/sql/notice_of_intent_submission/parcels/certificate_of_title/notice_of_intent_certificate_of_title.sql

@@ -0,0 +1,15 @@
+SELECT od.document_id,
+    od.document_code,
+    odc.description,
+    osp.subject_property_id,
+    oaa.alr_application_id,
+    od.who_created,
+    noid.uuid AS document_uuid
+FROM oats.oats_documents od
+    JOIN oats.oats_subject_properties osp ON osp.subject_property_id = od.subject_property_id
+    JOIN oats.oats_alr_applications oaa ON oaa.alr_application_id = osp.alr_application_id
+    JOIN oats.oats_document_codes odc ON odc.document_code = od.document_code
+    JOIN alcs.notice_of_intent_document noid ON noid.oats_document_id::bigint = od.document_id
+    JOIN alcs.notice_of_intent_parcel noip ON noip.oats_subject_property_id = osp.subject_property_id
+WHERE od.document_code = 'CT'
+    AND oaa.application_class_code = 'NOI'

bin/migrate-oats-data/noi/sql/notice_of_intent_submission/parcels/certificate_of_title/notice_of_intent_certificate_of_title_count.sql

@@ -0,0 +1,9 @@
+SELECT count(*)
+FROM oats.oats_documents od
+    JOIN oats.oats_subject_properties osp ON osp.subject_property_id = od.subject_property_id
+    JOIN oats.oats_alr_applications oaa ON oaa.alr_application_id = osp.alr_application_id
+    JOIN oats.oats_document_codes odc ON odc.document_code = od.document_code
+    JOIN alcs.notice_of_intent_document noid ON noid.oats_document_id::bigint = od.document_id
+    JOIN alcs.notice_of_intent_parcel noip ON noip.oats_subject_property_id = osp.subject_property_id
+WHERE od.document_code = 'CT'
+    AND oaa.application_class_code = 'NOI';

bin/migrate-oats-data/noi/sql/notice_of_intent_submission/parcels/certificate_of_title/notice_of_intent_certificate_of_title_validation.sql

@@ -0,0 +1,33 @@
+WITH oats_cert_of_titles AS (
+    SELECT od.document_id,
+        osp.subject_property_id,
+        oaa.alr_application_id
+    FROM oats.oats_documents od
+        JOIN oats.oats_subject_properties osp ON osp.subject_property_id = od.subject_property_id
+        JOIN oats.oats_alr_applications oaa ON oaa.alr_application_id = osp.alr_application_id
+        JOIN oats.oats_document_codes odc ON odc.document_code = od.document_code
+    WHERE od.document_code = 'CT'
+        AND oaa.application_class_code = 'NOI'
+),
+alcs_oats_certificate_of_titles AS (
+    SELECT noip.oats_subject_property_id,
+        noid.oats_document_id
+    FROM alcs.notice_of_intent_parcel noip
+        JOIN alcs.notice_of_intent_document noid ON noid.uuid = noip.certificate_of_title_uuid
+)
+SELECT *
+FROM alcs_oats_certificate_of_titles act
+    LEFT JOIN oats_cert_of_titles oct ON oct.subject_property_id = act.oats_subject_property_id
+WHERE act.oats_document_id != oct.document_id::TEXT;
+-- count certificates of title in oats linked to subject_properties
+SELECT count(*)
+FROM oats.oats_documents od
+    JOIN oats.oats_subject_properties osp ON osp.subject_property_id = od.subject_property_id
+    JOIN oats.oats_alr_applications oaa ON oaa.alr_application_id = osp.alr_application_id
+    JOIN oats.oats_document_codes odc ON odc.document_code = od.document_code
+WHERE od.document_code = 'CT'
+    AND oaa.application_class_code = 'NOI';
+-- count certificate of titles linked in alcs
+SELECT count(*)
+FROM alcs.notice_of_intent_parcel noip
+    JOIN alcs.notice_of_intent_document noid ON noid.uuid = noip.certificate_of_title_uuid;