From de125ec0c33783b8f23250afa773e90a8e84f625 Mon Sep 17 00:00:00 2001 From: ketaki-deodhar Date: Tue, 4 Aug 2026 07:22:48 -0700 Subject: [PATCH 1/6] 34055 - initial commit --- data-tool/flows/batch_delete_flow.py | 79 +++++++++++++++++++++++++++- 1 file changed, 77 insertions(+), 2 deletions(-) diff --git a/data-tool/flows/batch_delete_flow.py b/data-tool/flows/batch_delete_flow.py index 097e96a08f..aef39a8d51 100644 --- a/data-tool/flows/batch_delete_flow.py +++ b/data-tool/flows/batch_delete_flow.py @@ -15,14 +15,16 @@ businesses_cnt_query = """ SELECT COUNT(*) FROM businesses -WHERE 1 = 1 +WHERE 1 = 1 +AND identifier in ('BC1354075', 'BC1177462') AND legal_type IN ('BC', 'C', 'ULC', 'CUL', 'CC', 'CCC', 'QA', 'QB', 'QC', 'QD', 'QE', 'BEN', 'CP') AND legal_name LIKE '%' || :corp_name_suffix """ identifiers_query = """ SELECT id, identifier FROM businesses -WHERE 1 = 1 +WHERE 1 = 1 +AND identifier in ('BC1354075', 'BC1177462') AND legal_type IN ('BC', 'C', 'ULC', 'CUL', 'CC', 'CCC', 'QA', 'QB', 'QC', 'QD', 'QE', 'BEN', 'CP') AND legal_name LIKE '%' || :corp_name_suffix LIMIT :batch_size @@ -731,6 +733,76 @@ def get_selected_corps_mig(db_engine: Engine, config, candidates: List[str], off # nothing found in remaining pages return None, None, curr + +@task(cache_policy=NO_CACHE) +def get_lear_business_filings(db_engine: Engine, business_ids: List[int]): + """Return all filings for the provided business IDs where source = 'LEAR'.""" + if not business_ids: + return [] + + sql = text(""" + SELECT * + FROM filings + WHERE business_id IN :business_ids + AND source = :source + """).bindparams( + bindparam('business_ids', expanding=True), + bindparam('source'), + ) + + with db_engine.connect() as conn: + rows = conn.execute(sql, { + 'business_ids': business_ids, + 'source': 'LEAR', + }).fetchall() + + return [dict(row._mapping) for row in rows] + + +@task(cache_policy=NO_CACHE) +def insert_demigrated_filings(db_engine: Engine, business_ids: List[int], identifiers: List[str]): + """Fetch any LEAR filings for the provided businesses and persist them for demigration tracking.""" + if not business_ids: + return + + filings = get_lear_business_filings(db_engine, business_ids) + if not filings: + return + + id_to_identifier = dict(zip(business_ids, identifiers)) if business_ids and identifiers else {} + normalized_filings = [] + for filing in filings: + if not filing: + continue + + filing_dict = dict(filing) if hasattr(filing, 'keys') else filing + filing_id = filing_dict.get('id') + if filing_id is None: + continue + + business_id = filing_dict.get('business_id') + normalized_filings.append({ + 'identifier': id_to_identifier.get(business_id), + 'filing_id': filing_id, + 'business_id': business_id, + 'transaction_id': filing_dict.get('transaction_id'), + 'source': filing_dict.get('source') or 'LEAR', + }) + + if not normalized_filings: + return + + sql = text(""" + INSERT INTO demigrated_filings (corp_num, business_id, transaction_id, source) + VALUES (:identifier, :business_id, :transaction_id, :source) + ON CONFLICT (id) DO NOTHING + """) + + with db_engine.connect() as conn: + conn.execute(sql, normalized_filings) + conn.commit() + + @flow(log_prints=True) def batch_delete_flow(): try: @@ -773,6 +845,9 @@ def batch_delete_flow(): break print(f'🚀 Running round {cnt} to delete {len(business_ids)} busiesses...') + + insert_demigrated_filings(lear_engine, business_ids, identifiers) + futures = [] futures.append(lear_delete.submit(lear_engine, business_ids)) From ca0ff96db19aa7bd99aa8960e8572f24ff3b00fb Mon Sep 17 00:00:00 2001 From: ketaki-deodhar Date: Thu, 6 Aug 2026 12:42:35 -0700 Subject: [PATCH 2/6] 34055 - updates --- data-tool/flows/batch_delete_flow.py | 66 ++++++++++++++++++++++++---- data-tool/flows/config.py | 2 + 2 files changed, 60 insertions(+), 8 deletions(-) diff --git a/data-tool/flows/batch_delete_flow.py b/data-tool/flows/batch_delete_flow.py index aef39a8d51..340222b151 100644 --- a/data-tool/flows/batch_delete_flow.py +++ b/data-tool/flows/batch_delete_flow.py @@ -760,16 +760,22 @@ def get_lear_business_filings(db_engine: Engine, business_ids: List[int]): @task(cache_policy=NO_CACHE) -def insert_demigrated_filings(db_engine: Engine, business_ids: List[int], identifiers: List[str]): +def insert_demigrated_filings(lear_engine: Engine, colin_engine: Engine, business_ids: List[int], identifiers: List[str]): """Fetch any LEAR filings for the provided businesses and persist them for demigration tracking.""" if not business_ids: return - filings = get_lear_business_filings(db_engine, business_ids) + filings = get_lear_business_filings(lear_engine, business_ids) if not filings: return id_to_identifier = dict(zip(business_ids, identifiers)) if business_ids and identifiers else {} + + def _serialize_if_json(value): + if isinstance(value, (dict, list)): + return __import__('json').dumps(value) + return value + normalized_filings = [] for filing in filings: if not filing: @@ -784,21 +790,64 @@ def insert_demigrated_filings(db_engine: Engine, business_ids: List[int], identi normalized_filings.append({ 'identifier': id_to_identifier.get(business_id), 'filing_id': filing_id, - 'business_id': business_id, + 'filing_date': filing_dict.get('filing_date'), + 'filing_type': filing_dict.get('filing_type'), + 'filing_json': _serialize_if_json(filing_dict.get('filing_json')), + 'payment_id': filing_dict.get('payment_id'), 'transaction_id': filing_dict.get('transaction_id'), + 'business_id': business_id, + 'submitter_id': filing_dict.get('submitter_id'), + 'status': filing_dict.get('status'), + 'payment_completion_date': filing_dict.get('payment_completion_date'), + 'paper_only': filing_dict.get('paper_only'), + 'completion_date': filing_dict.get('completion_date'), + 'effective_date': filing_dict.get('effective_date'), 'source': filing_dict.get('source') or 'LEAR', + 'parent_filing_id': filing_dict.get('parent_filing_id'), + 'payment_status_code': filing_dict.get('payment_status_code'), + 'temp_reg': filing_dict.get('temp_reg'), + 'payment_account': filing_dict.get('payment_account'), + 'tech_correction_json': _serialize_if_json(filing_dict.get('tech_correction_json')), + 'colin_only': filing_dict.get('colin_only'), + 'deletion_locked': filing_dict.get('deletion_locked'), + 'order_details': _serialize_if_json(filing_dict.get('order_details')), + 'submitter_roles': _serialize_if_json(filing_dict.get('submitter_roles')), + 'meta_data': _serialize_if_json(filing_dict.get('meta_data')), + 'filing_sub_type': filing_dict.get('filing_sub_type'), + 'approval_type': filing_dict.get('approval_type'), + 'application_date': filing_dict.get('application_date'), + 'notice_date': filing_dict.get('notice_date'), + 'resubmission_date': filing_dict.get('resubmission_date'), + 'hide_in_ledger': filing_dict.get('hide_in_ledger'), + 'withdrawn_filing_id': filing_dict.get('withdrawn_filing_id'), + 'withdrawal_pending': filing_dict.get('withdrawal_pending'), + 'lear_only': filing_dict.get('lear_only') }) if not normalized_filings: return sql = text(""" - INSERT INTO demigrated_filings (corp_num, business_id, transaction_id, source) - VALUES (:identifier, :business_id, :transaction_id, :source) + INSERT INTO demigrated_filings (corp_num, filing_id, filing_date, filing_type, filing_json, + payment_id, transaction_id, business_id, submitter_id, status, + payment_completion_date, paper_only, completion_date, effective_date, + source, parent_filing_id, payment_status_code, temp_reg, payment_account, + tech_correction_json, colin_only, deletion_locked, order_details, + submitter_roles, meta_data, filing_sub_type, approval_type, + application_date, notice_date, resubmission_date, hide_in_ledger, + withdrawn_filing_id, withdrawal_pending, lear_only) + VALUES (:identifier, :filing_id, :filing_date, :filing_type, :filing_json, + :payment_id, :transaction_id, :business_id, :submitter_id, :status, + :payment_completion_date, :paper_only, :completion_date, :effective_date, + :source, :parent_filing_id, :payment_status_code, :temp_reg, :payment_account, + :tech_correction_json, :colin_only, :deletion_locked, :order_details, + :submitter_roles, :meta_data, :filing_sub_type, :approval_type, + :application_date, :notice_date, :resubmission_date, :hide_in_ledger, + :withdrawn_filing_id, :withdrawal_pending, :lear_only) ON CONFLICT (id) DO NOTHING """) - with db_engine.connect() as conn: + with colin_engine.connect() as conn: conn.execute(sql, normalized_filings) conn.commit() @@ -845,8 +894,9 @@ def batch_delete_flow(): break print(f'🚀 Running round {cnt} to delete {len(business_ids)} busiesses...') - - insert_demigrated_filings(lear_engine, business_ids, identifiers) + # If business has lear filings save them to demigrated filings table + if config.SAVE_DEMIGRATED_LEAR_FILINGS: + insert_demigrated_filings(lear_engine, colin_engine, business_ids, identifiers) futures = [] futures.append(lear_delete.submit(lear_engine, business_ids)) diff --git a/data-tool/flows/config.py b/data-tool/flows/config.py index bbad3b7e73..789db8733a 100644 --- a/data-tool/flows/config.py +++ b/data-tool/flows/config.py @@ -290,6 +290,8 @@ class _Config(): # pylint: disable=too-few-public-methods MIG_GROUP_IDS = os.getenv('MIG_GROUP_IDS') MIG_BATCH_IDS = os.getenv('MIG_BATCH_IDS') + SAVE_DEMIGRATED_LEAR_FILINGS = os.getenv('SAVE_DEMIGRATED_LEAR_FILINGS', 'False') == 'True' + # ------------------------------------------------------------------------------------------ # Auth-only flows (auth_processing tracking) # ------------------------------------------------------------------------------------------ From 25bfde1b04ae014a1d1631ce54a78b4fa183ffc9 Mon Sep 17 00:00:00 2001 From: ketaki-deodhar Date: Thu, 6 Aug 2026 12:45:04 -0700 Subject: [PATCH 3/6] 34055 - remove identofiers --- data-tool/flows/batch_delete_flow.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/data-tool/flows/batch_delete_flow.py b/data-tool/flows/batch_delete_flow.py index 340222b151..0989de418a 100644 --- a/data-tool/flows/batch_delete_flow.py +++ b/data-tool/flows/batch_delete_flow.py @@ -16,7 +16,6 @@ businesses_cnt_query = """ SELECT COUNT(*) FROM businesses WHERE 1 = 1 -AND identifier in ('BC1354075', 'BC1177462') AND legal_type IN ('BC', 'C', 'ULC', 'CUL', 'CC', 'CCC', 'QA', 'QB', 'QC', 'QD', 'QE', 'BEN', 'CP') AND legal_name LIKE '%' || :corp_name_suffix """ @@ -24,7 +23,6 @@ identifiers_query = """ SELECT id, identifier FROM businesses WHERE 1 = 1 -AND identifier in ('BC1354075', 'BC1177462') AND legal_type IN ('BC', 'C', 'ULC', 'CUL', 'CC', 'CCC', 'QA', 'QB', 'QC', 'QD', 'QE', 'BEN', 'CP') AND legal_name LIKE '%' || :corp_name_suffix LIMIT :batch_size From 1c2cb145568a2a7fe30b86fb835b268fcc960e83 Mon Sep 17 00:00:00 2001 From: ketaki-deodhar Date: Thu, 6 Aug 2026 12:52:22 -0700 Subject: [PATCH 4/6] 34055 - update query spaces --- data-tool/flows/batch_delete_flow.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/data-tool/flows/batch_delete_flow.py b/data-tool/flows/batch_delete_flow.py index 0989de418a..9eecd2af9b 100644 --- a/data-tool/flows/batch_delete_flow.py +++ b/data-tool/flows/batch_delete_flow.py @@ -15,14 +15,14 @@ businesses_cnt_query = """ SELECT COUNT(*) FROM businesses -WHERE 1 = 1 +WHERE 1 = 1 AND legal_type IN ('BC', 'C', 'ULC', 'CUL', 'CC', 'CCC', 'QA', 'QB', 'QC', 'QD', 'QE', 'BEN', 'CP') AND legal_name LIKE '%' || :corp_name_suffix """ identifiers_query = """ SELECT id, identifier FROM businesses -WHERE 1 = 1 +WHERE 1 = 1 AND legal_type IN ('BC', 'C', 'ULC', 'CUL', 'CC', 'CCC', 'QA', 'QB', 'QC', 'QD', 'QE', 'BEN', 'CP') AND legal_name LIKE '%' || :corp_name_suffix LIMIT :batch_size From d54a60da0c35dd6928051656b413eabcbc97cf91 Mon Sep 17 00:00:00 2001 From: ketaki-deodhar Date: Thu, 6 Aug 2026 13:45:00 -0700 Subject: [PATCH 5/6] 34055 - remove court order columns from demigrated filings table --- data-tool/scripts/colin_corps_extract_postgres_ddl | 4 ---- data-tool/tests/delta_restore/fixtures/minimal_schema.sql | 4 ---- 2 files changed, 8 deletions(-) diff --git a/data-tool/scripts/colin_corps_extract_postgres_ddl b/data-tool/scripts/colin_corps_extract_postgres_ddl index e2914f96e6..4450cbfb5d 100644 --- a/data-tool/scripts/colin_corps_extract_postgres_ddl +++ b/data-tool/scripts/colin_corps_extract_postgres_ddl @@ -1312,13 +1312,9 @@ create table if not exists demigrated_filings ( payment_status_code character varying(50), temp_reg character varying(10), payment_account character varying(30), - court_order_file_number character varying(20), - court_order_date timestamp with time zone, - court_order_effect_of_order character varying(500), tech_correction_json jsonb, colin_only boolean, deletion_locked boolean, - order_details character varying(2000), submitter_roles character varying(200), meta_data jsonb, filing_sub_type character varying(30), diff --git a/data-tool/tests/delta_restore/fixtures/minimal_schema.sql b/data-tool/tests/delta_restore/fixtures/minimal_schema.sql index 644a534faa..2706d2b55a 100644 --- a/data-tool/tests/delta_restore/fixtures/minimal_schema.sql +++ b/data-tool/tests/delta_restore/fixtures/minimal_schema.sql @@ -170,13 +170,9 @@ CREATE TABLE demigrated_filings ( payment_status_code text, temp_reg text, payment_account text, - court_order_file_number text, - court_order_date timestamptz, - court_order_effect_of_order text, tech_correction_json jsonb, colin_only boolean, deletion_locked boolean, - order_details text, submitter_roles text, meta_data jsonb, filing_sub_type text, From 92411ddd899fe976e2a085f491c76d57a059f722 Mon Sep 17 00:00:00 2001 From: ketaki-deodhar Date: Thu, 6 Aug 2026 15:18:48 -0700 Subject: [PATCH 6/6] 34055 - remove order details column ref --- data-tool/flows/batch_delete_flow.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/data-tool/flows/batch_delete_flow.py b/data-tool/flows/batch_delete_flow.py index 9eecd2af9b..93d9f9beae 100644 --- a/data-tool/flows/batch_delete_flow.py +++ b/data-tool/flows/batch_delete_flow.py @@ -808,7 +808,6 @@ def _serialize_if_json(value): 'tech_correction_json': _serialize_if_json(filing_dict.get('tech_correction_json')), 'colin_only': filing_dict.get('colin_only'), 'deletion_locked': filing_dict.get('deletion_locked'), - 'order_details': _serialize_if_json(filing_dict.get('order_details')), 'submitter_roles': _serialize_if_json(filing_dict.get('submitter_roles')), 'meta_data': _serialize_if_json(filing_dict.get('meta_data')), 'filing_sub_type': filing_dict.get('filing_sub_type'), @@ -830,7 +829,7 @@ def _serialize_if_json(value): payment_id, transaction_id, business_id, submitter_id, status, payment_completion_date, paper_only, completion_date, effective_date, source, parent_filing_id, payment_status_code, temp_reg, payment_account, - tech_correction_json, colin_only, deletion_locked, order_details, + tech_correction_json, colin_only, deletion_locked, submitter_roles, meta_data, filing_sub_type, approval_type, application_date, notice_date, resubmission_date, hide_in_ledger, withdrawn_filing_id, withdrawal_pending, lear_only) @@ -838,7 +837,7 @@ def _serialize_if_json(value): :payment_id, :transaction_id, :business_id, :submitter_id, :status, :payment_completion_date, :paper_only, :completion_date, :effective_date, :source, :parent_filing_id, :payment_status_code, :temp_reg, :payment_account, - :tech_correction_json, :colin_only, :deletion_locked, :order_details, + :tech_correction_json, :colin_only, :deletion_locked, :submitter_roles, :meta_data, :filing_sub_type, :approval_type, :application_date, :notice_date, :resubmission_date, :hide_in_ledger, :withdrawn_filing_id, :withdrawal_pending, :lear_only)