Refactor dlt in dag and model names (#227)

* update dlt script
* change progress bar to log for airflow
* remove target from dbt run in dag
* fix tag on dag and add info in example to find if change tracking is enabled
* Add instructions on how to use dbt-osmosis and update project with refactored descriptions and col types
* rename L1 models with stg_ prefit

Author: noel

load/dlt/loans_data.py

@@ -1,4 +1,4 @@
-#!/usr/bin/env -S uv run --verbose --cache-dir /tmp/.uv_cache
+#!/usr/bin/env -S uv run
 # /// script
 # dependencies = [
 #   "dlt[snowflake, parquet]==1.5.0",
@@ -28,7 +28,7 @@ def zip_coordinates():
 
 @dlt.source
 def loans_data():
-    return personal_loans, zip_coordinates
+    return [personal_loans, zip_coordinates]
 
 if __name__ == "__main__":
     datacoves_snowflake = dlt.destinations.snowflake(
@@ -46,8 +46,6 @@ def loans_data():
         dataset_name="loans"
     )
 
-    load_info = pipeline.run([
-            loans_data()
-        ])
+    load_info = pipeline.run(loans_data())
 
     print(load_info)

load/dlt/us_population.py

@@ -1,4 +1,4 @@
-#!/usr/bin/env -S uv run --cache-dir /tmp/.uv_cache
+#!/usr/bin/env -S uv run
 # /// script
 # dependencies = [
 #   "dlt[snowflake, parquet]==1.5.0",
@@ -30,7 +30,7 @@ def us_population_source():
     )
 
     pipeline = dlt.pipeline(
-        progress = "enlighten",
+        progress = "log",
         pipeline_name = "loans",
         destination = datacoves_snowflake,
         pipelines_dir = pipelines_dir,
@@ -40,7 +40,7 @@ def us_population_source():
     )
 
     load_info = pipeline.run([
-            us_population()
+            us_population_source()
         ])
 
     print(load_info)

observe/osmosis/.env.sample

@@ -0,0 +1 @@
+OPENAI_API_KEY=abc....

observe/osmosis/readme.md

@@ -0,0 +1,22 @@
+# dbt-osmosis
+
+## Set openai key from .env file
+`set -a && source /config/workspace/observe/osmosis/.env && set +a`
+
+## Run Refactor
+Executes organize which syncs yaml files with database schema.
+This also uses OpenAI to add descriptions to cols.
+
+`uvx --with=dbt-snowflake~=1.8.0 --from 'dbt-osmosis[openai]' dbt-osmosis yaml refactor --synthesize`
+
+Run without OpenAI
+
+`uvx --with=dbt-snowflake~=1.8.0 dbt-osmosis yaml refactor`
+
+You can pass `--fqn` with a name of a folder in `models/` to limit the refactor
+
+`dbt-osmosis yaml refactor --fqn L3_coves`
+
+Separate sub-folders with a `.`
+
+`dbt-osmosis yaml refactor --fqn L3_coves.loan_analytics`

orchestrate/dags/daily_loan_run.py

@@ -52,14 +52,26 @@ def extract_and_load_fivetran():
     @task_group(group_id="extract_and_load_dlt", tooltip="dlt Extract and Load")
     def extract_and_load_dlt():
         load_us_population = DatacovesBashOperator(
-            task_id="load_us_population",
-            bash_command="cd load/dlt && ./loans_data.py",
+            task_id="load_loads_data",
+            bash_command="""
+                cd load/dlt \
+                && ./loans_data.py
+            """,
+            env={
+                "UV_CACHE_DIR": "/tmp/uv_cache",
+                "EXTRACT__NEXT_ITEM_MODE":"fifo",
+                "EXTRACT__MAX_PARALLEL_ITEMS":"1",
+                "EXTRACT__WORKERS":"1",
+                "NORMALIZE__WORKERS":"1",
+                "LOAD__WORKERS":"1",
+            },
+            append_env=True,
         )
 
     tg_extract_and_load_dlt = extract_and_load_dlt()
     transform = DatacovesDbtOperator(
         task_id="transform",
-        bash_command="dbt build -s 'tag:daily_run_airbyte+ tag:daily_run_fivetran+ -t prd'",
+        bash_command="dbt build -s 'tag:daily_run_airbyte+ tag:daily_run_fivetran+'",
     )
     transform.set_upstream(
         [

orchestrate/dags/default_args_dag.py

@@ -0,0 +1,20 @@
+from airflow.decorators import dag
+from orchestrate.utils.default_args import default_args
+from operators.datacoves.dbt import DatacovesDbtOperator
+
+
+@dag(
+    default_args=default_args,
+    description="Daily dbt run",
+    schedule="0 12 * * *",
+    tags=["transform"],
+    catchup=False,
+
+)
+def default_args_dag():
+    run_dbt = DatacovesDbtOperator(
+        task_id="run_dbt", bash_command="dbt run -s country_codes"
+    )
+
+
+dag = default_args_dag()

orchestrate/utils/default_args.py

@@ -0,0 +1,11 @@
+# utils/default_args.py
+from datetime import datetime, timedelta
+
+default_args = {
+    'owner': '[email protected]',
+    'email_on_failure': False,
+    'email_on_retry': False,
+    'retries': 3,
+    'retry_delay': timedelta(minutes=5),
+    'start_date': datetime(2023, 12, 1),
+}

transform/.dbt_coves/config.yml

@@ -4,8 +4,8 @@ generate:
     # schemas: # List of schema names where to look for source tables
     #   - RAW
     sources_destination: "models/L1_inlets/{{schema}}/_{{schema}}.yml" # Where sources yml files will be generated
-    models_destination: "models/L1_inlets/{{schema}}/{{relation}}.sql" # Where models sql files will be generated
-    model_props_destination: "models/L1_inlets/{{schema}}/{{relation}}.yml" # Where models yml files will be generated
+    models_destination: "models/L1_inlets/{{schema}}/stg_{{relation}}.sql" # Where models sql files will be generated
+    model_props_destination: "models/L1_inlets/{{schema}}/stg_{{relation}}.yml" # Where models yml files will be generated
     update_strategy: update # Action to perform when a property file exists. Options: update, recreate, fail, ask
     templates_folder: ".dbt_coves/templates" # Folder where source generation jinja templates are located.
     flatten_json_fields: "no" # Action to perform when VARIANT / JSON field is encounted

transform/.dbt_coves/templates/staging_model_props.yml

@@ -1,7 +1,7 @@
 version: 2
 
 models:
-  - name: {{model}}
+  - name: stg_{{ model | lower }}
     description: ''
     columns:
 {%- for cols in nested.values() %}

transform/dbt_project.yml

@@ -43,6 +43,7 @@ seeds:
     +persist_docs:
         relation: true
         columns: true
+    +dbt-osmosis: "{model}.yml"
 
 snapshots:
     +enabled: "{{ True if target.database == 'balboa' else False }}"
@@ -56,6 +57,7 @@ snapshots:
 models:
     # elementary:
     #     +schema: "elementary"
+    +dbt-osmosis: "{model}.yml"
 
     balboa:
         +materialized: view
@@ -126,7 +128,7 @@ models:
         - "{{ dbt_snow_mask.apply_masking_policy('models') if target.name == 'prd_pii' }}"
         # This macro creates a non_versioned view of a versioned dbt model
         # Useful if you want to have a versioned mart but dont want BI tool to break if version
-        # is updated. 
+        # is updated.
         # - "{{ create_latest_version_view() }}"
 
     # Snowflake Defaults