Add support for Snowflake key pair and OAuth auth

This update introduces support for Snowflake key pair (JWT) and OAuth authentication in Airflow DAGs, dbt profiles, and environment setup scripts. The .env.example and profiles.yml files now document and support multiple authentication modes. Airflow DAGs and credential checks are updated to accept password, key pair, or OAuth token. Docker and local environment scripts are improved to handle secrets securely and remap key paths for local development. Minor fixes and clarifications are also included in Makefile, SQL, and test configs.

Author: wyang10

.gitignore

@@ -41,3 +41,7 @@ dbt_packages/
 # Great Expectations heavy artifacts
 great_expectations/uncommitted/validations/**
 great_expectations/uncommitted/data_docs/**
+
+# Local secrets (e.g. Snowflake key pair)
+secrets/*
+!secrets/README.md

Makefile

@@ -25,43 +25,43 @@ help: ## 显示可用命令
 # 环境初始化 / 本地 dbt
 # ---------------------------
 .PHONY: env
-env:  ## 一键初始化本地环境（source init_env.sh）
+env:  ## 初始化本地 dbt 环境（注意：不会持久化到你的当前终端）
 	@test -x "$(PROJECT_ROOT)/init_env.sh" || { echo "❌ 找不到 init_env.sh 或无执行权限"; exit 1; }
-	@source "$(PROJECT_ROOT)/init_env.sh"
+	@bash -lc 'source "$(PROJECT_ROOT)/init_env.sh"; echo ""; echo "提示：如需让 venv/环境变量留在当前终端，请手动运行：source ./init_env.sh"'
 
 .PHONY: dbt-debug
 dbt-debug: env ## dbt debug 自检
-	@cd "$(PROJECT_ROOT)/data_pipeline" && dbt debug
+	@bash -lc 'source "$(PROJECT_ROOT)/init_env.sh" >/dev/null 2>&1 || true; cd "$(PROJECT_ROOT)/data_pipeline" && dbt debug'
 
 .PHONY: dbt-parse
 dbt-parse: env ## dbt parse（解析项目）
-	@cd "$(PROJECT_ROOT)/data_pipeline" && dbt parse
+	@bash -lc 'source "$(PROJECT_ROOT)/init_env.sh" >/dev/null 2>&1 || true; cd "$(PROJECT_ROOT)/data_pipeline" && dbt parse'
 
 .PHONY: dbt-ls
 dbt-ls: env ## 列出所有模型
-	@cd "$(PROJECT_ROOT)/data_pipeline" && dbt ls
+	@bash -lc 'source "$(PROJECT_ROOT)/init_env.sh" >/dev/null 2>&1 || true; cd "$(PROJECT_ROOT)/data_pipeline" && dbt ls'
 
 .PHONY: dbt-ls-bronze dbt-ls-silver dbt-ls-gold
 dbt-ls-bronze: env ## 列出 bronze 模型
-	@cd "$(PROJECT_ROOT)/data_pipeline" && dbt ls --select 'path:models/bronze'
+	@bash -lc 'source "$(PROJECT_ROOT)/init_env.sh" >/dev/null 2>&1 || true; cd "$(PROJECT_ROOT)/data_pipeline" && dbt ls --select "path:models/bronze"'
 dbt-ls-silver: env ## 列出 silver 模型
-	@cd "$(PROJECT_ROOT)/data_pipeline" && dbt ls --select 'path:models/silver'
+	@bash -lc 'source "$(PROJECT_ROOT)/init_env.sh" >/dev/null 2>&1 || true; cd "$(PROJECT_ROOT)/data_pipeline" && dbt ls --select "path:models/silver"'
 dbt-ls-gold:   env ## 列出 gold 模型
-	@cd "$(PROJECT_ROOT)/data_pipeline" && dbt ls --select 'path:models/gold'
+	@bash -lc 'source "$(PROJECT_ROOT)/init_env.sh" >/dev/null 2>&1 || true; cd "$(PROJECT_ROOT)/data_pipeline" && dbt ls --select "path:models/gold"'
 
 .PHONY: dbt-run-bronze dbt-run-silver dbt-run-gold dbt-build
 dbt-run-bronze: env ## 仅跑 bronze
-	@cd "$(PROJECT_ROOT)/data_pipeline" && dbt run --select 'path:models/bronze'
+	@bash -lc 'source "$(PROJECT_ROOT)/init_env.sh" >/dev/null 2>&1 || true; cd "$(PROJECT_ROOT)/data_pipeline" && dbt run --select "path:models/bronze"'
 dbt-run-silver: env ## 仅跑 silver
-	@cd "$(PROJECT_ROOT)/data_pipeline" && dbt run --select 'path:models/silver'
+	@bash -lc 'source "$(PROJECT_ROOT)/init_env.sh" >/dev/null 2>&1 || true; cd "$(PROJECT_ROOT)/data_pipeline" && dbt run --select "path:models/silver"'
 dbt-run-gold:   env ## 仅跑 gold
-	@cd "$(PROJECT_ROOT)/data_pipeline" && dbt run --select 'path:models/gold'
+	@bash -lc 'source "$(PROJECT_ROOT)/init_env.sh" >/dev/null 2>&1 || true; cd "$(PROJECT_ROOT)/data_pipeline" && dbt run --select "path:models/gold"'
 dbt-build:      env ## 全量构建（含测试）
-	@cd "$(PROJECT_ROOT)/data_pipeline" && dbt build
+	@bash -lc 'source "$(PROJECT_ROOT)/init_env.sh" >/dev/null 2>&1 || true; cd "$(PROJECT_ROOT)/data_pipeline" && dbt build'
 
 .PHONY: dbt-docs
 dbt-docs: env ## 生成 + 本地预览文档（前台）
-	@cd "$(PROJECT_ROOT)/data_pipeline" && dbt docs generate && dbt docs serve
+	@bash -lc 'source "$(PROJECT_ROOT)/init_env.sh" >/dev/null 2>&1 || true; cd "$(PROJECT_ROOT)/data_pipeline" && dbt docs generate && dbt docs serve'
 
 # ---------------------------
 # Docker / Airflow 管理

airflow/.env.example

@@ -7,18 +7,25 @@
  # @Description: 这是默认设置,请设置`customMade`, 打开koroFileHeader查看配置 进行设置: https://github.com/OBKoro1/koro1FileHeader/wiki/%E9%85%8D%E7%BD%AE
 ### 
 # Snowflake for dbt CLI (example values)
-SNOWFLAKE_ACCOUNT=ywatlqy-hx58385
-SNOWFLAKE_USER=whyang
-SNOWFLAKE_PASSWORD=Yang890522890522
+# NOTE: Do NOT put real credentials in this file; copy to airflow/.env locally.
+SNOWFLAKE_ACCOUNT=your_account
+SNOWFLAKE_USER=your_user
+SNOWFLAKE_PASSWORD=your_password
 SNOWFLAKE_ROLE=dbt_role
-SNOWFLAKE_WAREHOUSE=dbt_wh
-SNOWFLAKE_DATABASE=dbt_db
-SNOWFLAKE_SCHEMA=dbt_schema
+SNOWFLAKE_WAREHOUSE=DBT_WH
+SNOWFLAKE_DATABASE=DBT_DB
+SNOWFLAKE_SCHEMA=DBT_SCHEMA
 DBT_TARGET=dev
 
+# If your Snowflake account enforces MFA, username/password auth may fail for
+# programmatic tools (dbt / Airflow). In that case, prefer key pair auth:
+# DBT_TARGET=dev_keypair
+# SNOWFLAKE_PRIVATE_KEY_PATH=/opt/airflow/secrets/snowflake_rsa_key.p8
+# SNOWFLAKE_PRIVATE_KEY_PASSPHRASE=your_passphrase   # optional
+
 # Optional: Airflow alerts (email on task failure)
 # Configure SMTP in Airflow if you want emails delivered.
-ALERT_EMAIL=i.whyang@gmail.com
+ALERT_EMAIL=you@example.com
 
 # Optional: Airflow secrets (recommended for non-local use)
 # Generate with: python -c 'import secrets,base64; print(base64.b64encode(secrets.token_bytes(32)).decode())'
@@ -27,10 +34,10 @@ AIRFLOW__WEBSERVER__SECRET_KEY=
 
 # Optional: Minimal SMTP example (Gmail via STARTTLS)
 # NOTE: For Gmail, use an app password, not your main password.
-AIRFLOW__SMTP__SMTP_HOST=smtp.gmail.com
-AIRFLOW__SMTP__SMTP_PORT=587
-AIRFLOW__SMTP__SMTP_STARTTLS=True
-AIRFLOW__SMTP__SMTP_SSL=False
-AIRFLOW__SMTP__SMTP_USER=whyang
-AIRFLOW__SMTP__SMTP_PASSWORD=fsnopcddwsuebfqn
-AIRFLOW__SMTP__SMTP_MAIL_FROM=s.wyang@live.com
+# AIRFLOW__SMTP__SMTP_HOST=smtp.gmail.com
+# AIRFLOW__SMTP__SMTP_PORT=587
+# AIRFLOW__SMTP__SMTP_STARTTLS=True
+# AIRFLOW__SMTP__SMTP_SSL=False
+# AIRFLOW__SMTP__SMTP_USER=you@gmail.com
+# AIRFLOW__SMTP__SMTP_PASSWORD=APP_PASSWORD_HERE
+# AIRFLOW__SMTP__SMTP_MAIL_FROM=you@gmail.com

airflow/dags/dbt_daily.py

@@ -24,6 +24,7 @@ def g(k):
 # 统一的 ENV（两条 DAG 完全一致）
 env_vars = {
     "DBT_PROFILES_DIR": "/opt/airflow/dbt",
+    "DBT_TARGET": g("DBT_TARGET") or "dev",
     # Traceability: pass query tag to Snowflake via dbt profile
     # Format: dag:task:run_id:queue:ts:try:run_type
     "DBT_QUERY_TAG": (
@@ -39,6 +40,9 @@ def g(k):
     "SNOWFLAKE_ACCOUNT": g("SNOWFLAKE_ACCOUNT"),
     "SNOWFLAKE_USER": g("SNOWFLAKE_USER"),
     "SNOWFLAKE_PASSWORD": g("SNOWFLAKE_PASSWORD"),
+    "SNOWFLAKE_PRIVATE_KEY_PATH": g("SNOWFLAKE_PRIVATE_KEY_PATH"),
+    "SNOWFLAKE_PRIVATE_KEY_PASSPHRASE": g("SNOWFLAKE_PRIVATE_KEY_PASSPHRASE"),
+    "SNOWFLAKE_OAUTH_TOKEN": g("SNOWFLAKE_OAUTH_TOKEN"),
     "SNOWFLAKE_ROLE": g("SNOWFLAKE_ROLE"),
     "SNOWFLAKE_DATABASE": g("SNOWFLAKE_DATABASE"),
     "SNOWFLAKE_WAREHOUSE": g("SNOWFLAKE_WAREHOUSE"),

airflow/dags/dbt_layered_pipeline.py

@@ -25,7 +25,7 @@
 # 通常与项目根同路径
 DBT_PROFILES_DIR = "/opt/airflow/dbt"
 # 可用 .env 或容器 ENV 指定
-DBT_TARGET = os.environ.get("DBT_TARGET", "prod")
+DBT_TARGET = os.environ.get("DBT_TARGET", "dev")
 # 使用 run+test 拆分以体现“质量闸门”
 USE_DBT_BUILD = False
 # 每日 03:00 UTC 触发
@@ -70,6 +70,9 @@ def _g(k: str, default: str = "") -> str:
     "SNOWFLAKE_ACCOUNT": _g("SNOWFLAKE_ACCOUNT"),
     "SNOWFLAKE_USER": _g("SNOWFLAKE_USER"),
     "SNOWFLAKE_PASSWORD": _g("SNOWFLAKE_PASSWORD"),
+    "SNOWFLAKE_PRIVATE_KEY_PATH": _g("SNOWFLAKE_PRIVATE_KEY_PATH"),
+    "SNOWFLAKE_PRIVATE_KEY_PASSPHRASE": _g("SNOWFLAKE_PRIVATE_KEY_PASSPHRASE"),
+    "SNOWFLAKE_OAUTH_TOKEN": _g("SNOWFLAKE_OAUTH_TOKEN"),
     "SNOWFLAKE_ROLE": _g("SNOWFLAKE_ROLE"),
     "SNOWFLAKE_DATABASE": _g("SNOWFLAKE_DATABASE"),
     "SNOWFLAKE_WAREHOUSE": _g("SNOWFLAKE_WAREHOUSE"),

airflow/dags/dbt_pipeline_dag.py

@@ -16,6 +16,7 @@ def g(k):
 # 与 dbt_daily 完全一致的 ENV
 env_vars = {
     "DBT_PROFILES_DIR": "/opt/airflow/dbt",
+    "DBT_TARGET": g("DBT_TARGET") or "dev",
     # dag:task:run_id:queue:ts:try:run_type
     "DBT_QUERY_TAG": (
         "{{ dag.dag_id }}:"
@@ -30,6 +31,9 @@ def g(k):
     "SNOWFLAKE_ACCOUNT": g("SNOWFLAKE_ACCOUNT"),
     "SNOWFLAKE_USER": g("SNOWFLAKE_USER"),
     "SNOWFLAKE_PASSWORD": g("SNOWFLAKE_PASSWORD"),
+    "SNOWFLAKE_PRIVATE_KEY_PATH": g("SNOWFLAKE_PRIVATE_KEY_PATH"),
+    "SNOWFLAKE_PRIVATE_KEY_PASSPHRASE": g("SNOWFLAKE_PRIVATE_KEY_PASSPHRASE"),
+    "SNOWFLAKE_OAUTH_TOKEN": g("SNOWFLAKE_OAUTH_TOKEN"),
     "SNOWFLAKE_ROLE": g("SNOWFLAKE_ROLE"),
     "SNOWFLAKE_DATABASE": g("SNOWFLAKE_DATABASE"),
     "SNOWFLAKE_WAREHOUSE": g("SNOWFLAKE_WAREHOUSE"),

airflow/dags/lib/creds.py

@@ -15,5 +15,9 @@ def ok(k: str) -> bool:
             "sample",
         }
 
-    required = ("SNOWFLAKE_ACCOUNT", "SNOWFLAKE_USER", "SNOWFLAKE_PASSWORD")
-    return all(ok(k) for k in required)
+    # Accept password auth OR key pair auth OR oauth token auth.
+    base = ("SNOWFLAKE_ACCOUNT", "SNOWFLAKE_USER")
+    password_ok = ok("SNOWFLAKE_PASSWORD")
+    keypair_ok = ok("SNOWFLAKE_PRIVATE_KEY_PATH") or ok("SNOWFLAKE_PRIVATE_KEY")
+    oauth_ok = ok("SNOWFLAKE_OAUTH_TOKEN")
+    return all(ok(k) for k in base) and (password_ok or keypair_ok or oauth_ok)

airflow/dags/serving/dbt_gold_consumer.py

@@ -20,13 +20,18 @@ def _g(k: str, default: str = "") -> str:
 
 ENV_VARS = {
     "DBT_PROFILES_DIR": "/opt/airflow/dbt",
+    "DBT_TARGET": _g("DBT_TARGET", "dev"),
     "SNOWFLAKE_ACCOUNT": _g("SNOWFLAKE_ACCOUNT"),
     "SNOWFLAKE_USER": _g("SNOWFLAKE_USER"),
     "SNOWFLAKE_PASSWORD": _g("SNOWFLAKE_PASSWORD"),
+    "SNOWFLAKE_PRIVATE_KEY_PATH": _g("SNOWFLAKE_PRIVATE_KEY_PATH"),
+    "SNOWFLAKE_PRIVATE_KEY_PASSPHRASE": _g("SNOWFLAKE_PRIVATE_KEY_PASSPHRASE"),
+    "SNOWFLAKE_OAUTH_TOKEN": _g("SNOWFLAKE_OAUTH_TOKEN"),
     "SNOWFLAKE_ROLE": _g("SNOWFLAKE_ROLE"),
     "SNOWFLAKE_DATABASE": _g("SNOWFLAKE_DATABASE"),
     "SNOWFLAKE_WAREHOUSE": _g("SNOWFLAKE_WAREHOUSE"),
     "SNOWFLAKE_SCHEMA": _g("SNOWFLAKE_SCHEMA"),
+    "PATH": _g("PATH", "/home/airflow/.local/bin:/usr/local/bin:/usr/bin:/bin"),
 }
 
 

airflow/docker-compose.yml

@@ -1,5 +1,10 @@
 version: "3.8"
 
+# NOTE:
+# - This file is a legacy single-folder compose (SQLite + SequentialExecutor).
+# - The main stack lives at repo root: `docker-compose.yml` (Postgres + LocalExecutor).
+# - Do not hardcode secrets here; use env vars or an env file.
+
 services:
   airflow-init:
     build: .
@@ -10,9 +15,8 @@ services:
     environment:
       AIRFLOW__CORE__EXECUTOR: SequentialExecutor
       AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: sqlite:////opt/airflow/airflow.db
-      AIRFLOW__WEBSERVER__SECRET_KEY: simple-secret
-      AIRFLOW__CORE__FERNET_KEY: TN6KCTBau36eJ3SLKN08_K_u5_sNQs-kDKUEdSuSq4o=
-      AIRFLOW__WEBSERVER__SECRET_KEY: 8b748b37f196ab9e7801c5f907ba129e32ff071b53064c960082346da52ebf1f6391b166d255d77ab54583ac15618e8a2f445b05637b3add2140242aca2a8559
+      AIRFLOW__CORE__FERNET_KEY: ${AIRFLOW__CORE__FERNET_KEY:-}
+      AIRFLOW__WEBSERVER__SECRET_KEY: ${AIRFLOW__WEBSERVER__SECRET_KEY:-dev-change-me}
 
       DBT_PROFILES_DIR: /opt/airflow/dbt
     volumes:
@@ -32,7 +36,7 @@ services:
     environment:
       AIRFLOW__CORE__EXECUTOR: SequentialExecutor
       AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: sqlite:////opt/airflow/airflow.db
-      AIRFLOW__WEBSERVER__SECRET_KEY: simple-secret
+      AIRFLOW__WEBSERVER__SECRET_KEY: ${AIRFLOW__WEBSERVER__SECRET_KEY:-dev-change-me}
       DBT_PROFILES_DIR: /opt/airflow/dbt                     # <<<
     volumes:
       - airflow_home:/opt/airflow

data_pipeline/macros/discounted_amount.sql

@@ -1,3 +1,3 @@
 {% macro discounted_amount(extended_price, discount) -%}
 ({{ extended_price }} * (1 - {{ discount }}))
-{%- endmacro %}dbt run -s int_order_items_summary
+{%- endmacro %}