✨ Add dag to begin to mint global ids

dags/non_program/dewrangle_id_minting/global_id_minting.py

@@ -0,0 +1,169 @@
+"""DAG to mint Global IDs in Dewrangle for a given PostgreSQL table with descriptors."""
+
+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.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
+    from airflow.hooks.base import BaseHook
+import psycopg2
+
+logger = logging.getLogger(__name__)
+
+
+with DAG(
+    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(2026, 6, 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",
+        ),
+        "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",
+            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=",
+                "T3JnYW5pemF0aW9uOmNsZWhibTF4ZjAwZTdpY2VzZjI0d2tlNHk=",
+            ],
+            values_display={
+                "T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=": "dff dev",
+                "T3JnYW5pemF0aW9uOmNsZHN4MzRrbjAwMTRnMGVzY3JndzUzYWQ=": "Kids First DRC",
+                "T3JnYW5pemF0aW9uOmNsZWhibTF4ZjAwZTdpY2VzZjI0d2tlNHk=": "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,
+        },
+    )
+
+    {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"]
+        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_flag }}",
+        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

pipeline_docs/guides/how_to_mint_global_ids.md

@@ -0,0 +1,207 @@
+# 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. 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
+   a specific study, the column `studyGlobalId` is also required.
+
+## 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 <qa|prod> \
+  --db dcc \
+  --organization_id <dewrangle organization id> \
+  --manifest /tmp/<descriptor_schema>_<descriptor_table>_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 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`
+
+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`.