✨ D3B-2194 guide on adding models to an airflow dag

dags/kids_first/dbt_bash.py

@@ -0,0 +1,60 @@
+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": ["[email protected]"],
+        "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,
+    )

dags/kids_first/example_study.py

@@ -0,0 +1,35 @@
+from airflow.sdk import Variable
+
+from cosmos import (
+    DbtDag,
+    ProjectConfig,
+    ProfileConfig,
+    ExecutionConfig,
+    RenderConfig,
+)
+from cosmos.profiles import PostgresUserPasswordProfileMapping
+
+profile_config = ProfileConfig(
+    profile_name=Variable.get("DBT_PROFILE_NAME"),
+    target_name="prd",
+    profile_mapping=PostgresUserPasswordProfileMapping(
+        conn_id="postgres_dev_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"],
+)

dbt_project/models/kids_first/example_study/_kf_example_study__models.yml

@@ -0,0 +1,45 @@
+---
+version: 2
+
+models:
+  - name: my_first_dbt_model
+    description: "A starter dbt model"
+    config:
+      meta:
+        study: kf_example_study
+    columns:
+      - name: id
+        description: "The primary key for this table"
+        data_tests:
+          - unique
+          - not_null
+
+  - name: my_second_dbt_model
+    description: "A starter dbt model"
+    config:
+      meta:
+        study: kf_example_study
+    columns:
+      - name: id
+        description: "The primary key for this table"
+        data_tests:
+          - unique
+          - not_null
+
+  - name: my_third_dbt_model
+    description: "a model that maps values in alphabet_groups to
+      `my_second_dbt_model`"
+    config:
+      meta:
+        study: kf_example_study
+    columns:
+      - name: id
+        description: "The primary key for this table"
+        data_tests:
+          #- unique
+          - not_null
+      - name: letter
+        description: "The letter in the alphabet"
+        data_tests:
+          - unique
+          - not_null

dbt_project/models/kids_first/example_study/my_first_dbt_model.sql

@@ -0,0 +1,27 @@
+
+/*
+    Welcome to your first dbt model!
+    Did you know that you can also configure models directly within SQL files?
+    This will override configurations stated in dbt_project.yml
+
+    Try changing "table" to "view" below
+*/
+
+{{ config(materialized='table', schema='kf_example') }}
+
+with source_data as (
+
+    select 1 as id
+    union all
+    select null as id
+
+)
+
+select *
+from source_data
+
+/*
+    Uncomment the line below to remove records with null `id` values
+*/
+
+where id is not null

dbt_project/models/kids_first/example_study/my_second_dbt_model.sql

@@ -0,0 +1,6 @@
+{{ config(materialized='table', schema='kf_example') }}
+-- Use the `ref` function to select from other models
+
+select *
+from {{ ref('my_first_dbt_model') }}
+where id = 1

dbt_project/models/kids_first/example_study/my_third_dbt_model.sql

@@ -0,0 +1,7 @@
+{{ config(materialized='table', schema='kf_example') }}
+
+select 
+  id,
+  letter
+from {{ ref('my_second_dbt_model') }}
+left join {{ ref('alphabet_grouping') }} on id = letter_grouping

dbt_project/profiles.yml

@@ -1,6 +1,6 @@
 ---
 include_dbt_sandbox:
-  target: ci
+  target: dev
   outputs:
     ci:
       type: postgres
@@ -20,3 +20,22 @@ include_dbt_sandbox:
       threads: 4
       type: postgres
       user: "{{ env_var('INCLUDEWAREHOUSE_USERNAME') }}"
+    qa:
+      type: postgres
+      host: "{{ env_var('INCLUDEWAREHOUSE_HOST') }}"
+      user: "{{ env_var('INCLUDEWAREHOUSE_SCV_USERNAME') }}"
+      password: "{{ env_var('INCLUDEWAREHOUSE_SCV_PASSWORD') }}"
+      port: 5432
+      dbname: postgres
+      schema: qa
+      threads: 4
+    prd:
+      type: postgres
+      host: "{{ env_var('INCLUDEWAREHOUSE_HOST') }}"
+      user: "{{ env_var('INCLUDEWAREHOUSE_SCV_USERNAME') }}"
+      password: "{{ env_var('INCLUDEWAREHOUSE_SCV_PASSWORD') }}"
+      port: 5432
+      dbname: postgres
+      schema: prd
+      threads: 4
+

dbt_project/seeds/_seeds.yml

@@ -0,0 +1,12 @@
+seeds:
+    - name: alphabet_grouping
+      config:
+        schema: kf
+      description: |
+        An example seed file that groups letters of the alphabet. This seed is
+        used by the kids first example study models.
+      columns:
+        - name: letter
+          description: A letter of the alphabet
+        - name: letter_grouping
+          description: The grouping that the letter belongs to

dbt_project/seeds/alphabet_grouping.csv

@@ -0,0 +1,27 @@
+letter,letter_grouping
+a,1
+b,1
+c,2
+d,2
+e,3
+f,3
+g,3
+h,4
+i,4
+j,4
+k,5
+l,5
+m,6
+n,6
+o,7
+p,7
+q,7
+r,8
+s,8
+t,8
+u,9
+v,9
+w,9
+x,10
+y,10
+z,10