diff --git a/.github/workflows/export-to-airflow.yml b/.github/workflows/export-to-airflow.yml index 3153517..17eec17 100644 --- a/.github/workflows/export-to-airflow.yml +++ b/.github/workflows/export-to-airflow.yml @@ -3,8 +3,6 @@ name: Export Repo to Airflow on: push: branches: [ "main" ] - pull_request: - branches: [ "main" ] permissions: contents: read diff --git a/dags/kids_first/dataservice_studies.py b/dags/kids_first/dataservice_studies.py deleted file mode 100644 index f9dd14d..0000000 --- a/dags/kids_first/dataservice_studies.py +++ /dev/null @@ -1,38 +0,0 @@ -from airflow.sdk import Variable - -from cosmos import ( - DbtDag, - ProjectConfig, - ProfileConfig, - ExecutionConfig, - RenderConfig, -) -from cosmos.profiles import PostgresUserPasswordProfileMapping - -profile_config = ProfileConfig( - # make sure target_name and profile_mapping align - profile_name=Variable.get("DBT_PROFILE_NAME"), - target_name="prd", - profile_mapping=PostgresUserPasswordProfileMapping( - conn_id="postgres_prd_svc", - profile_args={"schema": "prd"}, - ), -) - -example_study_dag = DbtDag( - project_config=ProjectConfig( - Variable.get("DBT_PROJECT_DIR"), - install_dbt_deps=True, - ), - profile_config=profile_config, - execution_config=ExecutionConfig( - dbt_executable_path=Variable.get("DBT_EXECUTABLE_PATH"), - ), - render_config=RenderConfig(select=["config.meta.study:kf_dataservice_study"]), - # normal dag parameters - schedule="@daily", - dag_id="kf_dataservice_studies", - tags=["POC", "Kids First"], -) - - diff --git a/dags/kids_first/dbt_bash.py b/dags/kids_first/dbt_bash.py deleted file mode 100644 index db14d3a..0000000 --- a/dags/kids_first/dbt_bash.py +++ /dev/null @@ -1,60 +0,0 @@ -from datetime import datetime, timedelta -from airflow.sdk import Variable -import os - -# The DAG object; we'll need this to instantiate a DAG -from airflow.models.dag import DAG - -# Operators; we need this to operate! -from airflow.operators.bash import BashOperator -from airflow.operators.python import PythonOperator - -with DAG( - "dbt_bash_status", - # These args will get passed on to each operator - # You can override them on a per-task basis during operator initialization - default_args={ - "depends_on_past": False, - "email": ["airflow@example.com"], - "email_on_failure": False, - "email_on_retry": False, - "retries": 1, - "retry_delay": timedelta(minutes=5), - # 'queue': 'bash_queue', - # 'pool': 'backfill', - # 'priority_weight': 10, - # 'end_date': datetime(2016, 1, 1), - # 'wait_for_downstream': False, - # 'sla': timedelta(hours=2), - # 'execution_timeout': timedelta(seconds=300), - # 'on_failure_callback': some_function, # or list of functions - # 'on_success_callback': some_other_function, # or list of functions - # 'on_retry_callback': another_function, # or list of functions - # 'sla_miss_callback': yet_another_function, # or list of functions - # 'on_skipped_callback': another_function, #or list of functions - # 'trigger_rule': 'all_success' - }, - description="Provide dbt information", - schedule=timedelta(days=1), - start_date=datetime(2021, 1, 1), - catchup=False, - tags=["POC"], -) as dag: - - # t1, t2 and t3 are examples of tasks created by instantiating operators - t1 = BashOperator( - task_id="dbt_version", - bash_command=f"{os.environ['AIRFLOW_HOME']}/dbt_venv/bin/dbt --version", - ) - - def check_for_warehouse_host_var(): - warehouse_host = Variable.get("INCLUDEWAREHOUSE_HOST", default=None) - if warehouse_host: - print("INCLUDEWAREHOUSE_HOST exists!") - else: - print("INCLUDEWAREHOUSE_HOST variable is not set.") - - t2 = PythonOperator( - task_id="check_for_warehouse_host_var", - python_callable=check_for_warehouse_host_var, - ) diff --git a/dags/kids_first/example_study.py b/dags/kids_first/example_study.py deleted file mode 100644 index d6af732..0000000 --- a/dags/kids_first/example_study.py +++ /dev/null @@ -1,36 +0,0 @@ -from airflow.sdk import Variable - -from cosmos import ( - DbtDag, - ProjectConfig, - ProfileConfig, - ExecutionConfig, - RenderConfig, -) -from cosmos.profiles import PostgresUserPasswordProfileMapping - -profile_config = ProfileConfig( - # make sure target_name and profile_mapping align - profile_name=Variable.get("DBT_PROFILE_NAME"), - target_name="prd", - profile_mapping=PostgresUserPasswordProfileMapping( - conn_id="postgres_prd_svc", - profile_args={"schema": "prd"}, - ), -) - -example_study_dag = DbtDag( - project_config=ProjectConfig( - Variable.get("DBT_PROJECT_DIR"), - install_dbt_deps=True, - ), - profile_config=profile_config, - execution_config=ExecutionConfig( - dbt_executable_path=Variable.get("DBT_EXECUTABLE_PATH"), - ), - render_config=RenderConfig(select=["config.meta.study:kf_example_study"]), - # normal dag parameters - schedule="@daily", - dag_id="kf_example_study", - tags=["POC", "Kids First"], -) diff --git a/dags/kids_first/kf_sd_1nns3k8v.py b/dags/kids_first/kf_sd_1nns3k8v.py deleted file mode 100644 index 4d86a13..0000000 --- a/dags/kids_first/kf_sd_1nns3k8v.py +++ /dev/null @@ -1,36 +0,0 @@ -from airflow.sdk import Variable - -from cosmos import ( - DbtDag, - ProjectConfig, - ProfileConfig, - ExecutionConfig, - RenderConfig, -) -from cosmos.profiles import PostgresUserPasswordProfileMapping - -profile_config = ProfileConfig( - # make sure target_name and profile_mapping align - profile_name=Variable.get("DBT_PROFILE_NAME"), - target_name="prd", - profile_mapping=PostgresUserPasswordProfileMapping( - conn_id="postgres_prd_svc", - profile_args={"schema": "prd"}, - ), -) - -example_study_dag = DbtDag( - project_config=ProjectConfig( - Variable.get("DBT_PROJECT_DIR"), - install_dbt_deps=True, - ), - profile_config=profile_config, - execution_config=ExecutionConfig( - dbt_executable_path=Variable.get("DBT_EXECUTABLE_PATH"), - ), - render_config=RenderConfig(select=["config.meta.study:kf_sd_1nns3k8v"]), - # normal dag parameters - schedule="@daily", - dag_id="kf_sd_1nns3k8v_dbt_dag", - tags=["POC", "Kids First"], -) \ No newline at end of file diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py deleted file mode 100644 index 0bc7cb7..0000000 --- a/dags/non_program/dewrangle_id_minting/global_id_minting.py +++ /dev/null @@ -1,169 +0,0 @@ -"""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