From e62d8337949b211470854536686faa44959f14ef Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Wed, 24 Jun 2026 09:46:04 -0400 Subject: [PATCH 01/30] =?UTF-8?q?=E2=9C=A8=20Add=20dag=20to=20begin=20to?= =?UTF-8?q?=20mint=20global=20ids?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../dewrangle_id_minting/global_id_minting.py | 86 +++++++++++++++++++ 1 file changed, 86 insertions(+) create mode 100644 dags/non_program/dewrangle_id_minting/global_id_minting.py diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py new file mode 100644 index 00000000..1268f886 --- /dev/null +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -0,0 +1,86 @@ +"""Boilerplate DAG that reads from a PostgreSQL table and saves to a local file.""" + +from datetime import datetime +import csv +import logging + +from airflow.models.dag import DAG +from airflow.models import Param +from airflow.operators.python import PythonOperator +from airflow.hooks.base import BaseHook +import psycopg2 + +logger = logging.getLogger(__name__) + + +with DAG( + dag_id="global_id_minting", + description="Read from PostgreSQL table and export to CSV file.", + start_date=datetime(2024, 1, 1), + schedule=None, + catchup=False, + params={ + "table_name": Param( + default="default_table", + type="string", + title="Table Name", + description="Name of the table to query from the database", + ), + "schema_name": Param( + default="default_schema", + type="string", + title="Schema Name", + description="Schema name where the table resides", + ), + }, + tags=["example", "postgres", "export"], +) as dag: + + def read_table_to_file(**context): + """Read from PostgreSQL table and save to local file.""" + schema_name = context["params"]["schema_name"] + table_name = context["params"]["table_name"] + + # Get Airflow connection + conn = BaseHook.get_connection("postgres_prd_svc") + + # Connect to PostgreSQL + connection = psycopg2.connect( + host=conn.host, + port=conn.port or 5432, + database=conn.schema, + user=conn.login, + password=conn.password, + ) + + cursor = connection.cursor() + + # Query the table + query = f"SELECT * FROM {schema_name}.{table_name}" + cursor.execute(query) + + # Get column names + column_names = [desc[0] for desc in cursor.description] + rows = cursor.fetchall() + + logger.info(f"Column names: {column_names}") + logger.info(f"Number of rows: {len(rows)}") + + # Save to local file + output_file = f"/tmp/{schema_name}_{table_name}_export.csv" + with open(output_file, "w", newline="") as f: + writer = csv.writer(f) + writer.writerow(column_names) + writer.writerows(rows) + + cursor.close() + connection.close() + + logger.info(f"Data exported to {output_file}") + return output_file + + read_and_export = PythonOperator( + task_id="read_and_export", + python_callable=read_table_to_file, + provide_context=True, + ) From cb12a59bba3defcb57e5f9e6b8428e07f4eb1895 Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Wed, 24 Jun 2026 10:29:38 -0400 Subject: [PATCH 02/30] =?UTF-8?q?=F0=9F=90=9B=20Remove=20provide=5Fcontext?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dags/non_program/dewrangle_id_minting/global_id_minting.py | 1 - 1 file changed, 1 deletion(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index 1268f886..06267116 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -82,5 +82,4 @@ def read_table_to_file(**context): read_and_export = PythonOperator( task_id="read_and_export", python_callable=read_table_to_file, - provide_context=True, ) From 67b05b1dbd966cc80f83b83b30951c63767ad2ee Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Thu, 25 Jun 2026 10:18:54 -0400 Subject: [PATCH 03/30] =?UTF-8?q?=F0=9F=9A=A7=20Add=20a=20step=20to=20use?= =?UTF-8?q?=20the=20output=20file?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../non_program/dewrangle_id_minting/global_id_minting.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index 06267116..e0f8051b 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -7,6 +7,7 @@ from airflow.models.dag import DAG from airflow.models import Param from airflow.operators.python import PythonOperator +from airflow.operators.bash import BashOperator from airflow.hooks.base import BaseHook import psycopg2 @@ -83,3 +84,10 @@ def read_table_to_file(**context): task_id="read_and_export", python_callable=read_table_to_file, ) + + wc_output_file = BashOperator( + task_id="wc_output_file", + bash_command="wc {{ ti.xcom_pull(task_ids='read_and_export') }}", + ) + + read_and_export >> wc_output_file From 93caf9580d908b076395ea1e4bb97adb666045fa Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Tue, 30 Jun 2026 15:09:25 -0400 Subject: [PATCH 04/30] =?UTF-8?q?=E2=9C=A8=20Add=20command=20for=20global?= =?UTF-8?q?=20id=20minting?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../dewrangle_id_minting/global_id_minting.py | 33 ++++++++++++++----- 1 file changed, 24 insertions(+), 9 deletions(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index e0f8051b..6931151a 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -21,20 +21,35 @@ schedule=None, catchup=False, params={ + "schema_name": Param( + default="default_schema", + type="string", + title="Schema Name", + description="Schema name where the table resides", + ), "table_name": Param( default="default_table", type="string", title="Table Name", description="Name of the table to query from the database", ), - "schema_name": Param( - default="default_schema", + "env": Param( + default="qa", type="string", - title="Schema Name", - description="Schema name where the table resides", + title="Environment", + description="Environment for the global ID minting command", + enum=["qa", "prod"], + ), + "dewrangle_organization_id": Param( + default="T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=", + type="string", + title="Dewrangle Organization ID", + description="Organization ID for the dewrangle global ID minting command", + enum=["T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=", "T3JnYW5pemF0aW9uOmNsZHN4MzRrbjAwMTRnMGVzY3JndzUzYWQ=", "org_id_2"] + values_display={"T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=": "dff dev", "T3JnYW5pemF0aW9uOmNsZHN4MzRrbjAwMTRnMGVzY3JndzUzYWQ=": "Kids First DRC", "org_id_2": "INCLUDE DCC"}, ), }, - tags=["example", "postgres", "export"], + tags=["non_program", "id_minting", "dewrangle"], ) as dag: def read_table_to_file(**context): @@ -85,9 +100,9 @@ def read_table_to_file(**context): python_callable=read_table_to_file, ) - wc_output_file = BashOperator( - task_id="wc_output_file", - bash_command="wc {{ ti.xcom_pull(task_ids='read_and_export') }}", + mint_ids = BashOperator( + task_id="mint_ids", + bash_command="d3b-dewrangle global-id-mint --env {{ params.env }} --db dcc --organization_id {{ params.dewrangle_organization_id }} --manifest {{ ti.xcom_pull(task_ids='read_and_export') }}", ) - read_and_export >> wc_output_file + read_and_export >> mint_ids From 1cfa68d50652b9fccbfe68372260aef7d7c3b04d Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Tue, 30 Jun 2026 15:57:42 -0400 Subject: [PATCH 05/30] =?UTF-8?q?=E2=9C=A8=20Get=20env=20vars=20correct?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../dewrangle_id_minting/global_id_minting.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index 6931151a..d462d518 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -103,6 +103,17 @@ def read_table_to_file(**context): mint_ids = BashOperator( task_id="mint_ids", bash_command="d3b-dewrangle global-id-mint --env {{ params.env }} --db dcc --organization_id {{ params.dewrangle_organization_id }} --manifest {{ ti.xcom_pull(task_ids='read_and_export') }}", + env={ + "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA":"", + "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"", + "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA":"", + "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"" + "DCC_WAREHOUSE_HOST":"", + "DCC_WAREHOUSE_PORT":"", + "DCC_WAREHOUSE_DB_NAME":"", + "DCC_WAREHOUSE_DB_USER":"", + "DCC_WAREHOUSE_DB_USER_PW": "" + } ) read_and_export >> mint_ids From b6c49bf4432114fb5cedc1e0c63d77f11ef665c6 Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Wed, 8 Jul 2026 11:36:34 -0400 Subject: [PATCH 06/30] =?UTF-8?q?=F0=9F=94=90=20Add=20use=20of=20connectio?= =?UTF-8?q?ns=20to=20query=20and=20load=20into=20warehouse?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../dewrangle_id_minting/global_id_minting.py | 42 ++++++++++++------- 1 file changed, 27 insertions(+), 15 deletions(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index d462d518..d120bf3e 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -21,17 +21,29 @@ schedule=None, catchup=False, params={ - "schema_name": Param( + "descriptor_schema_name": Param( default="default_schema", type="string", title="Schema Name", - description="Schema name where the table resides", + description="Schema name where the table with descriptors that need global IDs is located", ), - "table_name": Param( + "descriptor_table_name": Param( default="default_table", type="string", title="Table Name", - description="Name of the table to query from the database", + description="Name of the table with descriptors that need global IDs", + ), + "globalid_schema_name": Param( + default="default_schema", + type="string", + title="Schema Name", + description="Schema name where the table with generated global IDs is located", + ), + "globalid_table_name": Param( + default="default_table", + type="string", + title="Table Name", + description="Name of the table with generated global IDs", ), "env": Param( default="qa", @@ -54,8 +66,8 @@ def read_table_to_file(**context): """Read from PostgreSQL table and save to local file.""" - schema_name = context["params"]["schema_name"] - table_name = context["params"]["table_name"] + schema_name = context["params"]["descriptor_schema_name"] + table_name = context["params"]["descriptor_table_name"] # Get Airflow connection conn = BaseHook.get_connection("postgres_prd_svc") @@ -104,15 +116,15 @@ def read_table_to_file(**context): task_id="mint_ids", bash_command="d3b-dewrangle global-id-mint --env {{ params.env }} --db dcc --organization_id {{ params.dewrangle_organization_id }} --manifest {{ ti.xcom_pull(task_ids='read_and_export') }}", env={ - "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA":"", - "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"", - "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA":"", - "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"" - "DCC_WAREHOUSE_HOST":"", - "DCC_WAREHOUSE_PORT":"", - "DCC_WAREHOUSE_DB_NAME":"", - "DCC_WAREHOUSE_DB_USER":"", - "DCC_WAREHOUSE_DB_USER_PW": "" + "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA":"{{ params.globalid_schema_name }}", + "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"{{ params.globalid_table_name }}", + "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA":"{{ params.globalid_schema_name }}", + "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"{{ params.globalid_table_name }}", + "DCC_WAREHOUSE_HOST":"{{ conn.postgres_prd_svc.host }}", + "DCC_WAREHOUSE_PORT":"{{ conn.postgres_prd_svc.port }}", + "DCC_WAREHOUSE_DB_NAME":"{{ conn.postgres_prd_svc.schema }}", + "DCC_WAREHOUSE_DB_USER":"{{ conn.postgres_prd_svc.login }}", + "DCC_WAREHOUSE_DB_USER_PW": "{{ conn.postgres_prd_svc.password }}" } ) From 4728b7927622c4526c1e7d82d765c843517db4fb Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Wed, 8 Jul 2026 11:47:31 -0400 Subject: [PATCH 07/30] =?UTF-8?q?=F0=9F=90=9B=20Add=20a=20needed=20comma?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dags/non_program/dewrangle_id_minting/global_id_minting.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index d120bf3e..3c3378a3 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -16,7 +16,10 @@ with DAG( dag_id="global_id_minting", - description="Read from PostgreSQL table and export to CSV file.", + description=""" + Given a PostgreSQL table with descriptors, generate global IDs and load + those IDs into a target table. + """, start_date=datetime(2024, 1, 1), schedule=None, catchup=False, @@ -57,7 +60,7 @@ type="string", title="Dewrangle Organization ID", description="Organization ID for the dewrangle global ID minting command", - enum=["T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=", "T3JnYW5pemF0aW9uOmNsZHN4MzRrbjAwMTRnMGVzY3JndzUzYWQ=", "org_id_2"] + enum=["T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=", "T3JnYW5pemF0aW9uOmNsZHN4MzRrbjAwMTRnMGVzY3JndzUzYWQ=", "org_id_2"], values_display={"T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=": "dff dev", "T3JnYW5pemF0aW9uOmNsZHN4MzRrbjAwMTRnMGVzY3JndzUzYWQ=": "Kids First DRC", "org_id_2": "INCLUDE DCC"}, ), }, From 13bdcbb11cad7a4e3b3e78cabd3509126fcd520c Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Wed, 8 Jul 2026 13:27:09 -0400 Subject: [PATCH 08/30] =?UTF-8?q?=E2=9C=8F=EF=B8=8F=20Access=20connections?= =?UTF-8?q?=20more=20generally?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../dewrangle_id_minting/global_id_minting.py | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index 3c3378a3..ca11da5a 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -27,7 +27,7 @@ "descriptor_schema_name": Param( default="default_schema", type="string", - title="Schema Name", + title="Descriptor Schema Name", description="Schema name where the table with descriptors that need global IDs is located", ), "descriptor_table_name": Param( @@ -66,14 +66,15 @@ }, tags=["non_program", "id_minting", "dewrangle"], ) as dag: + + # Get Airflow connection + conn = BaseHook.get_connection("postgres_prd_svc") - def read_table_to_file(**context): + def read_table_to_file(conn=conn, **context): """Read from PostgreSQL table and save to local file.""" schema_name = context["params"]["descriptor_schema_name"] table_name = context["params"]["descriptor_table_name"] - # Get Airflow connection - conn = BaseHook.get_connection("postgres_prd_svc") # Connect to PostgreSQL connection = psycopg2.connect( @@ -123,11 +124,11 @@ def read_table_to_file(**context): "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"{{ params.globalid_table_name }}", "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA":"{{ params.globalid_schema_name }}", "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"{{ params.globalid_table_name }}", - "DCC_WAREHOUSE_HOST":"{{ conn.postgres_prd_svc.host }}", - "DCC_WAREHOUSE_PORT":"{{ conn.postgres_prd_svc.port }}", - "DCC_WAREHOUSE_DB_NAME":"{{ conn.postgres_prd_svc.schema }}", - "DCC_WAREHOUSE_DB_USER":"{{ conn.postgres_prd_svc.login }}", - "DCC_WAREHOUSE_DB_USER_PW": "{{ conn.postgres_prd_svc.password }}" + "DCC_WAREHOUSE_HOST":"{{ conn.host }}", + "DCC_WAREHOUSE_PORT":"{{ conn.port }}", + "DCC_WAREHOUSE_DB_NAME":"{{ conn.schema }}", + "DCC_WAREHOUSE_DB_USER":"{{ conn.login }}", + "DCC_WAREHOUSE_DB_USER_PW": "{{ conn.password }}" } ) From e71346a982abd40ca7adfe966a898a226e49c606 Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Wed, 8 Jul 2026 13:27:56 -0400 Subject: [PATCH 09/30] =?UTF-8?q?=E2=9C=8F=EF=B8=8F=20Just=20access=20conn?= =?UTF-8?q?=20directly?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../dewrangle_id_minting/global_id_minting.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index ca11da5a..5c739687 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -124,11 +124,11 @@ def read_table_to_file(conn=conn, **context): "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"{{ params.globalid_table_name }}", "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA":"{{ params.globalid_schema_name }}", "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"{{ params.globalid_table_name }}", - "DCC_WAREHOUSE_HOST":"{{ conn.host }}", - "DCC_WAREHOUSE_PORT":"{{ conn.port }}", - "DCC_WAREHOUSE_DB_NAME":"{{ conn.schema }}", - "DCC_WAREHOUSE_DB_USER":"{{ conn.login }}", - "DCC_WAREHOUSE_DB_USER_PW": "{{ conn.password }}" + "DCC_WAREHOUSE_HOST":conn.host, + "DCC_WAREHOUSE_PORT":conn.port or 5432, + "DCC_WAREHOUSE_DB_NAME":conn.schema, + "DCC_WAREHOUSE_DB_USER":conn.login, + "DCC_WAREHOUSE_DB_USER_PW": conn.password } ) From bef8b3ee559a24e9a71be21820a4a767c2a6e8ab Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Thu, 9 Jul 2026 09:38:35 -0400 Subject: [PATCH 10/30] =?UTF-8?q?=F0=9F=94=90=20Use=20sandbox=20instead=20?= =?UTF-8?q?of=20prd?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dags/non_program/dewrangle_id_minting/global_id_minting.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index 5c739687..566b616e 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -68,7 +68,7 @@ ) as dag: # Get Airflow connection - conn = BaseHook.get_connection("postgres_prd_svc") + conn = BaseHook.get_connection("postgres_dev_svc") def read_table_to_file(conn=conn, **context): """Read from PostgreSQL table and save to local file.""" From 59d3890dfe8b0956755e99991f2b0e5f7eb0003e Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Thu, 9 Jul 2026 10:16:51 -0400 Subject: [PATCH 11/30] =?UTF-8?q?=F0=9F=93=8C=20Handle=20airflow=20version?= =?UTF-8?q?=20issue?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dags/non_program/dewrangle_id_minting/global_id_minting.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index 566b616e..f7abcc15 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -8,7 +8,10 @@ from airflow.models import Param from airflow.operators.python import PythonOperator from airflow.operators.bash import BashOperator -from airflow.hooks.base import BaseHook +try: + from airflow.sdk.bases.hook import BaseHook +except ImportError: # Since Airflow 3.1, the BaseHook is in the airflow.sdk.bases.hook module + from airflow.hooks.base import BaseHook import psycopg2 logger = logging.getLogger(__name__) From 5ed40df489b6f731d9228e666563ff6a42e429e5 Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Thu, 9 Jul 2026 11:40:16 -0400 Subject: [PATCH 12/30] =?UTF-8?q?=F0=9F=93=9D=20Write=20guide=20on=20id=20?= =?UTF-8?q?minting?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../guides/how_to_mint_global_ids.md | 193 ++++++++++++++++++ 1 file changed, 193 insertions(+) create mode 100644 pipeline_docs/guides/how_to_mint_global_ids.md diff --git a/pipeline_docs/guides/how_to_mint_global_ids.md b/pipeline_docs/guides/how_to_mint_global_ids.md new file mode 100644 index 00000000..6fbd6d1f --- /dev/null +++ b/pipeline_docs/guides/how_to_mint_global_ids.md @@ -0,0 +1,193 @@ +# How to Mint Global Identifiers in Airflow + +This guide explains how to mint global identifiers with the Airflow DAG +`global_id_minting`. + +The DAG performs two steps: + +1. exports all rows from a source warehouse table to a temporary CSV file +2. runs `d3b-dewrangle global-id-mint` against that CSV and loads the minted + IDs into a target warehouse table + +Global ID's are generally minted at some point in harmonization of a study at +the discretion of the harmonizer. + +## Before you start + +Make sure all of the following are true before triggering the DAG: + +1. You can access the hosted Airflow instance. If not, complete the setup in + [connect-to-airflow.md](/Users/friedmanc1/Documents/include-dbt-sandbox/pipeline_docs/guides/connect-to-airflow.md). +2. The DAG `global_id_minting` is available in Airflow. +3. Your source table already exists in the warehouse. +4. You know which warehouse schema and table should receive the minted global + IDs. +5. Your source table columns match the descriptor format expected by + `d3b-dewrangle global-id-mint`. Specifically, the columns required are + `fhirResourceType`, `descriptor`, and `descriptorState`. + +## What the DAG does + +When you trigger the DAG, Airflow: + +1. Reads every column and row from + `{descriptor_schema_name}.{descriptor_table_name}`. +2. Writes the result to a temporary file in `/tmp` on the worker. +3. Runs: + +```bash +d3b-dewrangle global-id-mint \ + --env \ + --db dcc \ + --organization_id \ + --manifest /tmp/__export.csv +``` + +1. Loads the minted identifiers into the target table named by + `globalid_schema_name` and `globalid_table_name`. + +## Parameters to provide + +Trigger the DAG with the following parameters. + +### `descriptor_schema_name` + +The schema that contains the descriptor table to export. + +Example: + +```json +"descriptor_schema_name": "public" +``` + +### `descriptor_table_name` + +The table that contains the descriptors that need global identifiers. + +Example: + +```json +"descriptor_table_name": "study_subject_descriptors" +``` + +### `globalid_schema_name` + +The schema where the minted identifier table should be written. + +Example: + +```json +"globalid_schema_name": "access" +``` + +### `globalid_table_name` + +The destination table name for minted identifiers. + +Example: + +```json +"globalid_table_name": "subject_global_ids" +``` + +### `env` + +The runtime environment passed to `d3b-dewrangle`. + +Allowed values: + +1. `qa` +2. `prod` + +Example: + +```json +"env": "qa" +``` + +### `dewrangle_organization_id` + +The dewrangle organization identifier used for minting. + +The DAG currently exposes these labeled options in Airflow: + +1. `dff dev` +2. `Kids First DRC` +3. `INCLUDE DCC` + +Choose the organization that should own the minted identifiers. Note that these +options are human-readable pointers to the organization ID that dewrangle uses. + +## Triggering the DAG + +1. Open the hosted Airflow UI. +2. Search for the DAG `global_id_minting`. +3. Open the DAG details page. +4. Click `Trigger DAG`. +5. Replace the default parameters with your run configuration. +6. Start the run. + +Example parameter payload: + +```json +{ + "descriptor_schema_name": "public", + "descriptor_table_name": "study_subject_descriptors", + "globalid_schema_name": "access", + "globalid_table_name": "subject_global_ids", + "env": "qa", + "dewrangle_organization_id": "T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=" +} +``` + +## How to monitor the run + +The DAG has two tasks: + +1. `read_and_export` +2. `mint_ids` + +Use task logs to diagnose failures. + +### If `read_and_export` fails + +Check for: + +1. an incorrect `descriptor_schema_name` +2. an incorrect `descriptor_table_name` +3. missing warehouse permissions +4. descriptor data that cannot be exported cleanly + +### If `mint_ids` fails + +Check for: + +1. an invalid `dewrangle_organization_id` +2. a mismatch between your descriptor CSV columns and what + `d3b-dewrangle global-id-mint` expects +3. a target schema or table name that should not be used for the selected run +4. downstream dewrangle or warehouse connectivity issues + +## Verifying results + +After the DAG succeeds: + +1. query the target table `{globalid_schema_name}.{globalid_table_name}` +2. confirm the expected number of rows were written +3. validate that the minted global identifiers match the descriptors you + supplied + +## Operational notes + +The current DAG implementation has a few important behaviors to keep in mind: + +1. The source export uses `SELECT *`, so the full source table is exported. +2. The exported CSV is written to a temporary file on the Airflow worker. +3. The DAG reads warehouse credentials from the Airflow connection + `postgres_dev_svc`. +4. The `env` parameter changes the dewrangle CLI flag, but the DAG code still + sources its warehouse connection details from the same Airflow connection. + +If you need production minting behavior that differs from the current DAG +implementation, review the DAG configuration before running it with `env` set +to `prod`. From 3971a1de8e57ecdc890d88a75eaaaab9be730d9a Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Thu, 9 Jul 2026 13:11:18 -0400 Subject: [PATCH 13/30] =?UTF-8?q?=F0=9F=90=9B=20Specify=20conn?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dags/non_program/dewrangle_id_minting/global_id_minting.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index f7abcc15..32a9c8a1 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -117,6 +117,9 @@ def read_table_to_file(conn=conn, **context): read_and_export = PythonOperator( task_id="read_and_export", python_callable=read_table_to_file, + op_kwargs={ + "conn": conn, + }, ) mint_ids = BashOperator( From 3d438d8b31de9cb727ae0d9f32c5736d12362e5f Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Thu, 9 Jul 2026 13:16:28 -0400 Subject: [PATCH 14/30] =?UTF-8?q?=F0=9F=90=9B=20Need=20to=20connect=20to?= =?UTF-8?q?=20prd=20because=20dev=20is=20in=20a=20different=20vpc?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dags/non_program/dewrangle_id_minting/global_id_minting.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index 32a9c8a1..82052dca 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -71,7 +71,7 @@ ) as dag: # Get Airflow connection - conn = BaseHook.get_connection("postgres_dev_svc") + conn = BaseHook.get_connection("postgres_prd_svc") def read_table_to_file(conn=conn, **context): """Read from PostgreSQL table and save to local file.""" From 4d81eb88cd4a54945b58ea18cb5b1e2743aaed4d Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Thu, 9 Jul 2026 13:20:34 -0400 Subject: [PATCH 15/30] =?UTF-8?q?=F0=9F=90=9B=20Can't=20pass=20an=20int?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dags/non_program/dewrangle_id_minting/global_id_minting.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index 82052dca..7e76c93b 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -131,7 +131,7 @@ def read_table_to_file(conn=conn, **context): "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA":"{{ params.globalid_schema_name }}", "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"{{ params.globalid_table_name }}", "DCC_WAREHOUSE_HOST":conn.host, - "DCC_WAREHOUSE_PORT":conn.port or 5432, + "DCC_WAREHOUSE_PORT":conn.port or "5432", "DCC_WAREHOUSE_DB_NAME":conn.schema, "DCC_WAREHOUSE_DB_USER":conn.login, "DCC_WAREHOUSE_DB_USER_PW": conn.password From 4ddb660171c83febb5d5d380402253909f0afd3f Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Fri, 10 Jul 2026 09:44:46 -0400 Subject: [PATCH 16/30] =?UTF-8?q?=F0=9F=90=9B=20Try=20setting=20port=20to?= =?UTF-8?q?=20a=20strig?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dags/non_program/dewrangle_id_minting/global_id_minting.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index 7e76c93b..b7ec5a61 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -131,7 +131,7 @@ def read_table_to_file(conn=conn, **context): "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA":"{{ params.globalid_schema_name }}", "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"{{ params.globalid_table_name }}", "DCC_WAREHOUSE_HOST":conn.host, - "DCC_WAREHOUSE_PORT":conn.port or "5432", + "DCC_WAREHOUSE_PORT":"5432", "DCC_WAREHOUSE_DB_NAME":conn.schema, "DCC_WAREHOUSE_DB_USER":conn.login, "DCC_WAREHOUSE_DB_USER_PW": conn.password From d6969878366c95a67c7eade91fb11471322122fd Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Fri, 10 Jul 2026 09:50:55 -0400 Subject: [PATCH 17/30] =?UTF-8?q?=F0=9F=93=9D=20Add=20instructions=20for?= =?UTF-8?q?=20minting=20intra-study=20id's?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipeline_docs/guides/how_to_mint_global_ids.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pipeline_docs/guides/how_to_mint_global_ids.md b/pipeline_docs/guides/how_to_mint_global_ids.md index 6fbd6d1f..5399131c 100644 --- a/pipeline_docs/guides/how_to_mint_global_ids.md +++ b/pipeline_docs/guides/how_to_mint_global_ids.md @@ -24,7 +24,8 @@ Make sure all of the following are true before triggering the DAG: IDs. 5. Your source table columns match the descriptor format expected by `d3b-dewrangle global-id-mint`. Specifically, the columns required are - `fhirResourceType`, `descriptor`, and `descriptorState`. + `fhirResourceType`, `descriptor`, and `descriptorState`. To mint IDs within + a specific study, the column `studyGlobalId` is also required. ## What the DAG does From 502a2c1f6feb4513e38a7d8da495236fc84e7f98 Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Fri, 10 Jul 2026 14:24:09 -0400 Subject: [PATCH 18/30] =?UTF-8?q?=F0=9F=90=9B=20Specify=20correct=20locati?= =?UTF-8?q?on=20of=20the=20id=20minting=20tool?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dags/non_program/dewrangle_id_minting/global_id_minting.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index b7ec5a61..c0e505ea 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -124,7 +124,7 @@ def read_table_to_file(conn=conn, **context): mint_ids = BashOperator( task_id="mint_ids", - bash_command="d3b-dewrangle global-id-mint --env {{ params.env }} --db dcc --organization_id {{ params.dewrangle_organization_id }} --manifest {{ ti.xcom_pull(task_ids='read_and_export') }}", + bash_command="${ID_MINTING_PATH}/bin/d3b-dewrangle global-id-mint --env {{ params.env }} --db dcc --organization_id {{ params.dewrangle_organization_id }} --manifest {{ ti.xcom_pull(task_ids='read_and_export') }}", env={ "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA":"{{ params.globalid_schema_name }}", "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"{{ params.globalid_table_name }}", From f28fd38057e140fec8fc21f5ecceaab16e4f78eb Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Fri, 10 Jul 2026 14:32:19 -0400 Subject: [PATCH 19/30] =?UTF-8?q?=F0=9F=90=9B=20Correct=20var=20name?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dags/non_program/dewrangle_id_minting/global_id_minting.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index c0e505ea..761ccd7b 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -124,7 +124,7 @@ def read_table_to_file(conn=conn, **context): mint_ids = BashOperator( task_id="mint_ids", - bash_command="${ID_MINTING_PATH}/bin/d3b-dewrangle global-id-mint --env {{ params.env }} --db dcc --organization_id {{ params.dewrangle_organization_id }} --manifest {{ ti.xcom_pull(task_ids='read_and_export') }}", + bash_command="${ID_MINTING_PATH}/bin/d3b-dewrangle global-id-mint --env {{ params.env }} --db dcc --organization-id {{ params.dewrangle_organization_id }} --manifest {{ ti.xcom_pull(task_ids='read_and_export') }}", env={ "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA":"{{ params.globalid_schema_name }}", "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"{{ params.globalid_table_name }}", From b26295b96ebbeadb854af8c629dbe38a5bbb3abd Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Fri, 10 Jul 2026 14:59:52 -0400 Subject: [PATCH 20/30] =?UTF-8?q?=F0=9F=94=90=20Use=20dewrangle=20token?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../dewrangle_id_minting/global_id_minting.py | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index 761ccd7b..c583c6dc 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -71,9 +71,10 @@ ) as dag: # Get Airflow connection - conn = BaseHook.get_connection("postgres_prd_svc") + postgres_conn = BaseHook.get_connection("postgres_prd_svc") + dewrangle_conn = BaseHook.get_connection("dewrangle_api") - def read_table_to_file(conn=conn, **context): + def read_table_to_file(conn=postgres_conn, **context): """Read from PostgreSQL table and save to local file.""" schema_name = context["params"]["descriptor_schema_name"] table_name = context["params"]["descriptor_table_name"] @@ -118,7 +119,7 @@ def read_table_to_file(conn=conn, **context): task_id="read_and_export", python_callable=read_table_to_file, op_kwargs={ - "conn": conn, + "conn": postgres_conn, }, ) @@ -130,11 +131,13 @@ def read_table_to_file(conn=conn, **context): "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"{{ params.globalid_table_name }}", "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA":"{{ params.globalid_schema_name }}", "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"{{ params.globalid_table_name }}", - "DCC_WAREHOUSE_HOST":conn.host, - "DCC_WAREHOUSE_PORT":"5432", - "DCC_WAREHOUSE_DB_NAME":conn.schema, - "DCC_WAREHOUSE_DB_USER":conn.login, - "DCC_WAREHOUSE_DB_USER_PW": conn.password + "DCC_WAREHOUSE_HOST":postgres_conn.host, + "DCC_WAREHOUSE_PORT":postgres_conn.port or 5432, + "DCC_WAREHOUSE_DB_NAME":postgres_conn.schema, + "DCC_WAREHOUSE_DB_USER":postgres_conn.login, + "DCC_WAREHOUSE_DB_USER_PW": postgres_conn.password, + "DEWRANGLE_BASE_URL":dewrangle_conn.host, + "DEWRANGLE_TOKEN":dewrangle_conn.password, } ) From 631e8fcc367f649bf58d84e8073b70672c1a5db5 Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Fri, 10 Jul 2026 15:03:07 -0400 Subject: [PATCH 21/30] =?UTF-8?q?=F0=9F=90=9B=20Port=20needs=20to=20be=20a?= =?UTF-8?q?=20string?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dags/non_program/dewrangle_id_minting/global_id_minting.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index c583c6dc..52dd1aca 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -132,7 +132,7 @@ def read_table_to_file(conn=postgres_conn, **context): "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA":"{{ params.globalid_schema_name }}", "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"{{ params.globalid_table_name }}", "DCC_WAREHOUSE_HOST":postgres_conn.host, - "DCC_WAREHOUSE_PORT":postgres_conn.port or 5432, + "DCC_WAREHOUSE_PORT":str(postgres_conn.port or 5432), "DCC_WAREHOUSE_DB_NAME":postgres_conn.schema, "DCC_WAREHOUSE_DB_USER":postgres_conn.login, "DCC_WAREHOUSE_DB_USER_PW": postgres_conn.password, From 39608f036252057c27efe0bacfebda7ffc5df892 Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Mon, 13 Jul 2026 11:41:53 -0400 Subject: [PATCH 22/30] =?UTF-8?q?=F0=9F=94=A7=20Set=20timeout?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dags/non_program/dewrangle_id_minting/global_id_minting.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index 52dd1aca..2c9b362e 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -36,19 +36,19 @@ "descriptor_table_name": Param( default="default_table", type="string", - title="Table Name", + title="Descriptor Table Name", description="Name of the table with descriptors that need global IDs", ), "globalid_schema_name": Param( default="default_schema", type="string", - title="Schema Name", + title="Global ID Schema Name", description="Schema name where the table with generated global IDs is located", ), "globalid_table_name": Param( default="default_table", type="string", - title="Table Name", + title="Global ID Table Name", description="Name of the table with generated global IDs", ), "env": Param( @@ -138,6 +138,7 @@ def read_table_to_file(conn=postgres_conn, **context): "DCC_WAREHOUSE_DB_USER_PW": postgres_conn.password, "DEWRANGLE_BASE_URL":dewrangle_conn.host, "DEWRANGLE_TOKEN":dewrangle_conn.password, + "DEWRANGLE_CLIENT_EXECUTION_TIMEOUT": "300", } ) From 87e9ff2473735be49438baaa1d2399770e54f4c6 Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Mon, 13 Jul 2026 13:53:48 -0400 Subject: [PATCH 23/30] =?UTF-8?q?=F0=9F=94=A7=20Set=20execution=20timeout?= =?UTF-8?q?=20as=20int?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dags/non_program/dewrangle_id_minting/global_id_minting.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index 2c9b362e..f2ee945b 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -138,7 +138,7 @@ def read_table_to_file(conn=postgres_conn, **context): "DCC_WAREHOUSE_DB_USER_PW": postgres_conn.password, "DEWRANGLE_BASE_URL":dewrangle_conn.host, "DEWRANGLE_TOKEN":dewrangle_conn.password, - "DEWRANGLE_CLIENT_EXECUTION_TIMEOUT": "300", + "DEWRANGLE_CLIENT_EXECUTION_TIMEOUT": 300, } ) From 195d278bfb4025c5dedc9b819cc28c25d064b5b7 Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Mon, 13 Jul 2026 14:40:09 -0400 Subject: [PATCH 24/30] =?UTF-8?q?=F0=9F=94=A7=20Timeout=20must=20be=20stri?= =?UTF-8?q?ng?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dags/non_program/dewrangle_id_minting/global_id_minting.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index f2ee945b..2c9b362e 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -138,7 +138,7 @@ def read_table_to_file(conn=postgres_conn, **context): "DCC_WAREHOUSE_DB_USER_PW": postgres_conn.password, "DEWRANGLE_BASE_URL":dewrangle_conn.host, "DEWRANGLE_TOKEN":dewrangle_conn.password, - "DEWRANGLE_CLIENT_EXECUTION_TIMEOUT": 300, + "DEWRANGLE_CLIENT_EXECUTION_TIMEOUT": "300", } ) From 6d1399323eb823882388531bcd9ac28916843cba Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Thu, 16 Jul 2026 11:32:50 -0400 Subject: [PATCH 25/30] =?UTF-8?q?=F0=9F=90=9B=20Create=20table?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../dewrangle_id_minting/global_id_minting.py | 252 +++++++++--------- 1 file changed, 131 insertions(+), 121 deletions(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index 2c9b362e..c2bea7e4 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -8,9 +8,12 @@ from airflow.models import Param from airflow.operators.python import PythonOperator from airflow.operators.bash import BashOperator + try: from airflow.sdk.bases.hook import BaseHook -except ImportError: # Since Airflow 3.1, the BaseHook is in the airflow.sdk.bases.hook module +except ( + ImportError +): # Since Airflow 3.1, the BaseHook is in the airflow.sdk.bases.hook module from airflow.hooks.base import BaseHook import psycopg2 @@ -18,128 +21,135 @@ with DAG( - dag_id="global_id_minting", - description=""" + dag_id="global_id_minting", + description=""" Given a PostgreSQL table with descriptors, generate global IDs and load those IDs into a target table. """, - start_date=datetime(2024, 1, 1), - schedule=None, - catchup=False, - params={ - "descriptor_schema_name": Param( - default="default_schema", - type="string", - title="Descriptor Schema Name", - description="Schema name where the table with descriptors that need global IDs is located", - ), - "descriptor_table_name": Param( - default="default_table", - type="string", - title="Descriptor Table Name", - description="Name of the table with descriptors that need global IDs", - ), - "globalid_schema_name": Param( - default="default_schema", - type="string", - title="Global ID Schema Name", - description="Schema name where the table with generated global IDs is located", - ), - "globalid_table_name": Param( - default="default_table", - type="string", - title="Global ID Table Name", - description="Name of the table with generated global IDs", - ), - "env": Param( - default="qa", - type="string", - title="Environment", - description="Environment for the global ID minting command", + start_date=datetime(2024, 1, 1), + schedule=None, + catchup=False, + params={ + "descriptor_schema_name": Param( + default="default_schema", + type="string", + title="Descriptor Schema Name", + description="Schema name where the table with descriptors that need global IDs is located", + ), + "descriptor_table_name": Param( + default="default_table", + type="string", + title="Descriptor Table Name", + description="Name of the table with descriptors that need global IDs", + ), + "globalid_schema_name": Param( + default="default_schema", + type="string", + title="Global ID Schema Name", + description="Schema name where the table with generated global IDs is located", + ), + "globalid_table_name": Param( + default="default_table", + type="string", + title="Global ID Table Name", + description="Name of the table with generated global IDs", + ), + "env": Param( + default="qa", + type="string", + title="Environment", + description="Environment for the global ID minting command", enum=["qa", "prod"], - ), - "dewrangle_organization_id": Param( - default="T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=", - type="string", - title="Dewrangle Organization ID", - description="Organization ID for the dewrangle global ID minting command", - enum=["T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=", "T3JnYW5pemF0aW9uOmNsZHN4MzRrbjAwMTRnMGVzY3JndzUzYWQ=", "org_id_2"], - values_display={"T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=": "dff dev", "T3JnYW5pemF0aW9uOmNsZHN4MzRrbjAwMTRnMGVzY3JndzUzYWQ=": "Kids First DRC", "org_id_2": "INCLUDE DCC"}, - ), - }, - tags=["non_program", "id_minting", "dewrangle"], + ), + "dewrangle_organization_id": Param( + default="T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=", + type="string", + title="Dewrangle Organization ID", + description="Organization ID for the dewrangle global ID minting command", + enum=[ + "T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=", + "T3JnYW5pemF0aW9uOmNsZHN4MzRrbjAwMTRnMGVzY3JndzUzYWQ=", + "org_id_2", + ], + values_display={ + "T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=": "dff dev", + "T3JnYW5pemF0aW9uOmNsZHN4MzRrbjAwMTRnMGVzY3JndzUzYWQ=": "Kids First DRC", + "org_id_2": "INCLUDE DCC", + }, + ), + }, + tags=["non_program", "id_minting", "dewrangle"], ) as dag: - - # Get Airflow connection - postgres_conn = BaseHook.get_connection("postgres_prd_svc") - dewrangle_conn = BaseHook.get_connection("dewrangle_api") - - def read_table_to_file(conn=postgres_conn, **context): - """Read from PostgreSQL table and save to local file.""" - schema_name = context["params"]["descriptor_schema_name"] - table_name = context["params"]["descriptor_table_name"] - - - # Connect to PostgreSQL - connection = psycopg2.connect( - host=conn.host, - port=conn.port or 5432, - database=conn.schema, - user=conn.login, - password=conn.password, - ) - - cursor = connection.cursor() - - # Query the table - query = f"SELECT * FROM {schema_name}.{table_name}" - cursor.execute(query) - - # Get column names - column_names = [desc[0] for desc in cursor.description] - rows = cursor.fetchall() - - logger.info(f"Column names: {column_names}") - logger.info(f"Number of rows: {len(rows)}") - - # Save to local file - output_file = f"/tmp/{schema_name}_{table_name}_export.csv" - with open(output_file, "w", newline="") as f: - writer = csv.writer(f) - writer.writerow(column_names) - writer.writerows(rows) - - cursor.close() - connection.close() - - logger.info(f"Data exported to {output_file}") - return output_file - - read_and_export = PythonOperator( - task_id="read_and_export", - python_callable=read_table_to_file, - op_kwargs={ - "conn": postgres_conn, - }, - ) - - mint_ids = BashOperator( - task_id="mint_ids", - bash_command="${ID_MINTING_PATH}/bin/d3b-dewrangle global-id-mint --env {{ params.env }} --db dcc --organization-id {{ params.dewrangle_organization_id }} --manifest {{ ti.xcom_pull(task_ids='read_and_export') }}", - env={ - "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA":"{{ params.globalid_schema_name }}", - "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"{{ params.globalid_table_name }}", - "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA":"{{ params.globalid_schema_name }}", - "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE":"{{ params.globalid_table_name }}", - "DCC_WAREHOUSE_HOST":postgres_conn.host, - "DCC_WAREHOUSE_PORT":str(postgres_conn.port or 5432), - "DCC_WAREHOUSE_DB_NAME":postgres_conn.schema, - "DCC_WAREHOUSE_DB_USER":postgres_conn.login, - "DCC_WAREHOUSE_DB_USER_PW": postgres_conn.password, - "DEWRANGLE_BASE_URL":dewrangle_conn.host, - "DEWRANGLE_TOKEN":dewrangle_conn.password, - "DEWRANGLE_CLIENT_EXECUTION_TIMEOUT": "300", - } - ) - - read_and_export >> mint_ids + + # Get Airflow connection + postgres_conn = BaseHook.get_connection("postgres_prd_svc") + dewrangle_conn = BaseHook.get_connection("dewrangle_api") + + def read_table_to_file(conn=postgres_conn, **context): + """Read from PostgreSQL table and save to local file.""" + schema_name = context["params"]["descriptor_schema_name"] + table_name = context["params"]["descriptor_table_name"] + + # Connect to PostgreSQL + connection = psycopg2.connect( + host=conn.host, + port=conn.port or 5432, + database=conn.schema, + user=conn.login, + password=conn.password, + ) + + cursor = connection.cursor() + + # Query the table + query = f"SELECT * FROM {schema_name}.{table_name}" + cursor.execute(query) + + # Get column names + column_names = [desc[0] for desc in cursor.description] + rows = cursor.fetchall() + + logger.info(f"Column names: {column_names}") + logger.info(f"Number of rows: {len(rows)}") + + # Save to local file + output_file = f"/tmp/{schema_name}_{table_name}_export.csv" + with open(output_file, "w", newline="") as f: + writer = csv.writer(f) + writer.writerow(column_names) + writer.writerows(rows) + + cursor.close() + connection.close() + + logger.info(f"Data exported to {output_file}") + return output_file + + read_and_export = PythonOperator( + task_id="read_and_export", + python_callable=read_table_to_file, + op_kwargs={ + "conn": postgres_conn, + }, + ) + + mint_ids = BashOperator( + task_id="mint_ids", + bash_command="${ID_MINTING_PATH}/bin/d3b-dewrangle global-id-mint --env {{ params.env }} --db dcc --organization-id {{ params.dewrangle_organization_id }} --manifest {{ ti.xcom_pull(task_ids='read_and_export') --create-dewrangle-ids-table}}", + env={ + "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA": "{{ params.globalid_schema_name }}", + "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE": "{{ params.globalid_table_name }}", + "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA": "{{ params.globalid_schema_name }}", + "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE": "{{ params.globalid_table_name }}", + "DCC_WAREHOUSE_HOST": postgres_conn.host, + "DCC_WAREHOUSE_PORT": str(postgres_conn.port or 5432), + "DCC_WAREHOUSE_DB_NAME": postgres_conn.schema, + "DCC_WAREHOUSE_DB_USER": postgres_conn.login, + "DCC_WAREHOUSE_DB_USER_PW": postgres_conn.password, + "DEWRANGLE_BASE_URL": dewrangle_conn.host, + "DEWRANGLE_TOKEN": dewrangle_conn.password, + "DEWRANGLE_CLIENT_EXECUTION_TIMEOUT": "300", + }, + ) + + read_and_export >> mint_ids From cd60ac4f82bc34060e45c2af069b7dcf4bc06e7f Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Thu, 16 Jul 2026 11:41:34 -0400 Subject: [PATCH 26/30] =?UTF-8?q?=F0=9F=90=9B=20Put=20comman=20arg=20in=20?= =?UTF-8?q?correct=20place?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dags/non_program/dewrangle_id_minting/global_id_minting.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index c2bea7e4..7af1a5e9 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -135,7 +135,7 @@ def read_table_to_file(conn=postgres_conn, **context): mint_ids = BashOperator( task_id="mint_ids", - bash_command="${ID_MINTING_PATH}/bin/d3b-dewrangle global-id-mint --env {{ params.env }} --db dcc --organization-id {{ params.dewrangle_organization_id }} --manifest {{ ti.xcom_pull(task_ids='read_and_export') --create-dewrangle-ids-table}}", + bash_command="${ID_MINTING_PATH}/bin/d3b-dewrangle global-id-mint --env {{ params.env }} --db dcc --organization-id {{ params.dewrangle_organization_id }} --manifest {{ ti.xcom_pull(task_ids='read_and_export') }} --create-dewrangle-ids-table", env={ "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA": "{{ params.globalid_schema_name }}", "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE": "{{ params.globalid_table_name }}", From 29180e8bf1e621a609b1d14dd353e252d2c70639 Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Wed, 22 Jul 2026 09:15:40 -0400 Subject: [PATCH 27/30] =?UTF-8?q?=F0=9F=8E=A8=20Doc=20updates?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../dewrangle_id_minting/global_id_minting.py | 4 ++-- pipeline_docs/guides/how_to_mint_global_ids.md | 15 ++++++++++++++- 2 files changed, 16 insertions(+), 3 deletions(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index 7af1a5e9..04d3e570 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -1,4 +1,4 @@ -"""Boilerplate DAG that reads from a PostgreSQL table and saves to a local file.""" +"""DAG to mint Global IDs in Dewrangle for a given PostgreSQL table with descriptors.""" from datetime import datetime import csv @@ -26,7 +26,7 @@ Given a PostgreSQL table with descriptors, generate global IDs and load those IDs into a target table. """, - start_date=datetime(2024, 1, 1), + start_date=datetime(2026, 6, 1), schedule=None, catchup=False, params={ diff --git a/pipeline_docs/guides/how_to_mint_global_ids.md b/pipeline_docs/guides/how_to_mint_global_ids.md index 5399131c..7c4c59a9 100644 --- a/pipeline_docs/guides/how_to_mint_global_ids.md +++ b/pipeline_docs/guides/how_to_mint_global_ids.md @@ -49,7 +49,20 @@ d3b-dewrangle global-id-mint \ ## Parameters to provide -Trigger the DAG with the following parameters. +Trigger the DAG with the following parameters below. While there is the option +to enter each parameter individually, airflow allows entering these parameters +as a JSON object: + +```json +{ + "env": "qa", + "globalid_table_name": "global_ids", + "globalid_schema_name": "friedmanc1_dev_schema", + "descriptor_table_name": "dewrangle_id_minting_test", + "descriptor_schema_name": "friedmanc1_dev_schema", + "dewrangle_organization_id": "T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=" +} +``` ### `descriptor_schema_name` From ad34df3c2a65d2a0cddd4132d4d76d40f8bf8e42 Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Thu, 23 Jul 2026 10:09:16 -0400 Subject: [PATCH 28/30] =?UTF-8?q?=F0=9F=93=9D=20Add=20note=20re:=20target?= =?UTF-8?q?=20table=20not=20needing=20to=20exist?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipeline_docs/guides/how_to_mint_global_ids.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pipeline_docs/guides/how_to_mint_global_ids.md b/pipeline_docs/guides/how_to_mint_global_ids.md index 7c4c59a9..1d5180d5 100644 --- a/pipeline_docs/guides/how_to_mint_global_ids.md +++ b/pipeline_docs/guides/how_to_mint_global_ids.md @@ -21,7 +21,7 @@ Make sure all of the following are true before triggering the DAG: 2. The DAG `global_id_minting` is available in Airflow. 3. Your source table already exists in the warehouse. 4. You know which warehouse schema and table should receive the minted global - IDs. + IDs. Note that this table does not need to exist. 5. Your source table columns match the descriptor format expected by `d3b-dewrangle global-id-mint`. Specifically, the columns required are `fhirResourceType`, `descriptor`, and `descriptorState`. To mint IDs within From c8965cb45cc9c63ed87d9289030e39e90dc1933c Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Fri, 24 Jul 2026 09:44:23 -0400 Subject: [PATCH 29/30] =?UTF-8?q?=E2=9C=A8=20Add=20flag=20to=20create=20ne?= =?UTF-8?q?w=20global=20id=20table?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../dewrangle_id_minting/global_id_minting.py | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index 04d3e570..a1dea211 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -54,6 +54,12 @@ title="Global ID Table Name", description="Name of the table with generated global IDs", ), + "create_new_globalid_table": Param( + default=False, + type="boolean", + title="Create New Global ID Table", + description="Whether to create a new table for the generated global IDs or not", + ), "env": Param( default="qa", type="string", @@ -133,9 +139,15 @@ def read_table_to_file(conn=postgres_conn, **context): }, ) + create_dewrangle_ids_table_flag = ( + "--create-dewrangle-ids-table" + if dag.params["create_new_globalid_table"] + else "" + ) + mint_ids = BashOperator( task_id="mint_ids", - bash_command="${ID_MINTING_PATH}/bin/d3b-dewrangle global-id-mint --env {{ params.env }} --db dcc --organization-id {{ params.dewrangle_organization_id }} --manifest {{ ti.xcom_pull(task_ids='read_and_export') }} --create-dewrangle-ids-table", + bash_command="${ID_MINTING_PATH}/bin/d3b-dewrangle global-id-mint --env {{ params.env }} --db dcc --organization-id {{ params.dewrangle_organization_id }} --manifest {{ ti.xcom_pull(task_ids='read_and_export') }} {{ create_dewrangle_ids_table_flag }}", env={ "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA": "{{ params.globalid_schema_name }}", "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE": "{{ params.globalid_table_name }}", From be91582ef601f55552fd5d75fa30a954d7fdac2f Mon Sep 17 00:00:00 2001 From: chris-s-friedman Date: Mon, 27 Jul 2026 14:49:04 -0400 Subject: [PATCH 30/30] =?UTF-8?q?=F0=9F=9A=80=20Add=20include=20dewrangle?= =?UTF-8?q?=20org=20ID?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dags/non_program/dewrangle_id_minting/global_id_minting.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py index a1dea211..0bc7cb70 100644 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -75,12 +75,12 @@ enum=[ "T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=", "T3JnYW5pemF0aW9uOmNsZHN4MzRrbjAwMTRnMGVzY3JndzUzYWQ=", - "org_id_2", + "T3JnYW5pemF0aW9uOmNsZWhibTF4ZjAwZTdpY2VzZjI0d2tlNHk=", ], values_display={ "T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=": "dff dev", "T3JnYW5pemF0aW9uOmNsZHN4MzRrbjAwMTRnMGVzY3JndzUzYWQ=": "Kids First DRC", - "org_id_2": "INCLUDE DCC", + "T3JnYW5pemF0aW9uOmNsZWhibTF4ZjAwZTdpY2VzZjI0d2tlNHk=": "INCLUDE DCC", }, ), }, @@ -139,6 +139,8 @@ def read_table_to_file(conn=postgres_conn, **context): }, ) + {set create_dewrangle_ids_table_flag = "--create-dewrangle-ids-table" if dag.params["create_new_globalid_table"] else ""} + create_dewrangle_ids_table_flag = ( "--create-dewrangle-ids-table" if dag.params["create_new_globalid_table"]