🚀 Deployment & CI/CD — Ship It!

my_project:
  target: dev                    # Default target when you run dbt locally
  outputs:

    dev:
      type: postgres
      host: localhost
      user: analytics_dev
      password: "{{ env_var('DBT_DEV_PASSWORD') }}"
      database: warehouse
      schema: dev_analytics          # ← Your test kitchen
      threads: 4

    prod:
      type: postgres
      host: prod-warehouse.company.com
      user: analytics_prod
      password: "{{ env_var('DBT_PROD_PASSWORD') }}"
      database: warehouse
      schema: prod_analytics         # ← The restaurant floor
      threads: 8

SELECT
    order_id,
    customer_id,
    order_date,
    amount
FROM
    {{ ref('stg_orders') }}

{% if target.name == 'dev' %}
    -- In dev: only process last 7 days (fast iteration!)
    WHERE order_date >= CURRENT_DATE - INTERVAL '7 days'
{% endif %}

-- In prod: this WHERE clause is skipped → processes ALL data

# Step 1: Load reference data (country codes, status mappings, etc.)
$ dbt seed

# Step 2: Build all models (staging → intermediate → marts)
$ dbt run

# Step 3: Run ALL tests (schema tests, custom tests, freshness)
$ dbt test

# Or combine steps 2 & 3 with "build" (runs + tests in DAG order)
$ dbt build

name: dbt CI

on:
  pull_request:
    branches: [main]

jobs:
  dbt-test:
    runs-on: ubuntu-latest

    steps:
      - name: Checkout code
        uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install dbt
        run: pip install dbt-postgres

      - name: Install packages
        run: dbt deps

      - name: Run modified models
        run: dbt run --select state:modified+
        env:
          DBT_PROFILES_DIR: .
          DBT_PASSWORD: ${{ secrets.DBT_PASSWORD }}

      - name: Test modified models
        run: dbt test --select state:modified+
        env:
          DBT_PROFILES_DIR: .
          DBT_PASSWORD: ${{ secrets.DBT_PASSWORD }}

# Only run models that changed since the last production run
$ dbt run --select state:modified+ --defer --state ./prod-artifacts/

# state:modified  → models YOU changed in this PR
# +               → plus everything DOWNSTREAM of those models
# --defer         → for unchanged models, use prod tables instead of rebuilding
# --state         → path to production's manifest.json (the "before" snapshot)

name: dbt Deploy to Production

on:
  push:
    branches: [main]

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - run: pip install dbt-postgres
      - run: dbt deps
      - run: dbt seed --target prod
      - run: dbt run --target prod
      - run: dbt test --target prod
        env:
          DBT_PROFILES_DIR: .
          DBT_PROD_PASSWORD: ${{ secrets.DBT_PROD_PASSWORD }}

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
    'email_on_failure': True,
    'email': ['data-alerts@company.com'],
}

with DAG(
    dag_id='dbt_production_pipeline',
    default_args=default_args,
    schedule_interval='0 6 * * *',     # Every day at 6 AM
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:

    dbt_seed = BashOperator(
        task_id='dbt_seed',
        bash_command='cd /opt/dbt/my_project && dbt seed --target prod',
    )

    dbt_run = BashOperator(
        task_id='dbt_run',
        bash_command='cd /opt/dbt/my_project && dbt run --target prod',
    )

    dbt_test = BashOperator(
        task_id='dbt_test',
        bash_command='cd /opt/dbt/my_project && dbt test --target prod',
    )

    notify_success = BashOperator(
        task_id='notify_success',
        bash_command='echo "dbt pipeline completed successfully!"',
    )

    # Define the execution order
    dbt_seed >> dbt_run >> dbt_test >> notify_success

from cosmos import DbtDag, ProjectConfig, ProfileConfig

dbt_dag = DbtDag(
    project_config=ProjectConfig("/opt/dbt/my_project"),
    profile_config=ProfileConfig(
        profile_name="my_project",
        target_name="prod",
    ),
    schedule_interval="0 6 * * *",
    start_date=datetime(2024, 1, 1),
    dag_id="dbt_cosmos_pipeline",
)

# Cosmos auto-generates one Airflow task per dbt model!
# stg_customers → stg_orders → fct_orders → dim_customers
# Each model is a separate, retryable task in Airflow.

my_project:
  target: prod
  outputs:
    prod:
      type: snowflake
      account: "{{ env_var('SNOWFLAKE_ACCOUNT') }}"
      user: "{{ env_var('SNOWFLAKE_USER') }}"
      password: "{{ env_var('SNOWFLAKE_PASSWORD') }}"
      role: "{{ env_var('SNOWFLAKE_ROLE') }}"
      database: "{{ env_var('SNOWFLAKE_DATABASE') }}"
      warehouse: "{{ env_var('SNOWFLAKE_WAREHOUSE') }}"
      schema: prod_analytics
      threads: 8

# In dbt Cloud → Account Settings → Notifications

# Email notifications:
# ✅ On job failure → data-team@company.com
# ✅ On test failure → data-quality@company.com

# Slack notifications (via webhook):
# ✅ On job failure → #data-alerts channel
# ✅ On job success → #data-pipeline-log channel

# For GitHub Actions, add a Slack step:
- name: Notify Slack on failure
  if: failure()
  uses: slackapi/slack-github-action@v1
  with:
    channel-id: 'C01234ABCDE'
    slack-message: '🚨 dbt pipeline failed! Check: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}'
  env:
    SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}

# Step 1: Build everything in a staging schema
$ dbt run --target prod --vars '{"schema_suffix": "_staging"}'

# Step 2: Run ALL tests against the staging schema
$ dbt test --target prod --vars '{"schema_suffix": "_staging"}'

# Step 3: If all tests pass, swap the schemas (atomic operation)
ALTER SCHEMA prod_analytics RENAME TO prod_analytics_old;
ALTER SCHEMA prod_analytics_staging RENAME TO prod_analytics;

# Step 4: Drop the old schema (optional, keep as backup)
-- DROP SCHEMA prod_analytics_old CASCADE;

# Result: Zero downtime! Dashboards never saw stale or partial data.