dbt (Data Build Tool) & Data Quality

# Installation de dbt

installation = '''
# Adapter selon ton Data Warehouse/Lakehouse :

pip install dbt-core              # Core (obligatoire)

# + UN adapter selon ta cible :
pip install dbt-postgres          # PostgreSQL
pip install dbt-snowflake         # Snowflake
pip install dbt-bigquery          # BigQuery
pip install dbt-spark             # Spark (Databricks, EMR, local)
pip install dbt-duckdb            # DuckDB (local, léger)
pip install dbt-trino             # Trino/Presto

# Vérifier l'installation
dbt --version
'''

print(installation)
print("\n💡 Pour ce module, on utilise dbt-duckdb (léger, local) ou dbt-spark.")

# Créer un nouveau projet dbt

init_project = '''
# Créer un nouveau projet
dbt init my_dbt_project

# Structure créée :
my_dbt_project/
│
├── dbt_project.yml          # Configuration principale du projet
├── profiles.yml             # Connexions aux bases (souvent dans ~/.dbt/)
│
├── models/                  # ⭐ TES TRANSFORMATIONS SQL
│   └── example/
│       ├── my_first_model.sql
│       └── schema.yml       # Tests et documentation
│
├── seeds/                   # Données statiques (CSV)
│
├── snapshots/               # Historisation (SCD Type 2)
│
├── macros/                  # Fonctions Jinja réutilisables
│
├── tests/                   # Tests SQL custom
│
├── analyses/                # Requêtes ad-hoc (non matérialisées)
│
└── target/                  # Fichiers compilés (généré)
'''

print(init_project)

# dbt_project.yml - Configuration principale

dbt_project_yml = '''
# dbt_project.yml

name: 'ecommerce_analytics'      # Nom du projet (sans espaces, sans tirets)
version: '1.0.0'
config-version: 2

# Profil de connexion (défini dans profiles.yml)
profile: 'ecommerce_analytics'

# Chemins des différents composants
model-paths: ["models"]
analysis-paths: ["analyses"]
test-paths: ["tests"]
seed-paths: ["seeds"]
macro-paths: ["macros"]
snapshot-paths: ["snapshots"]
target-path: "target"           # Où dbt écrit les fichiers compilés
clean-targets: ["target", "dbt_packages"]

# ═══════════════════════════════════════════════════════════════
# Configuration des modèles (TRÈS IMPORTANT)
# ═══════════════════════════════════════════════════════════════

models:
  ecommerce_analytics:           # Doit correspondre au "name" ci-dessus
    
    # Configuration par défaut pour TOUS les modèles
    +materialized: view          # Par défaut : créer des vues
    
    # Configuration par dossier
    staging:                     # Dossier models/staging/
      +materialized: view
      +schema: staging           # Écrire dans le schéma "staging"
    
    silver:
      +materialized: table       # Tables complètes
      +schema: silver
    
    gold:
      +materialized: incremental # Incrémental pour la performance
      +schema: gold
'''

print(dbt_project_yml)

# profiles.yml - Connexions aux bases de données

profiles_yml = '''
# ~/.dbt/profiles.yml

ecommerce_analytics:             # Doit correspondre au "profile" dans dbt_project.yml
  
  target: dev                    # Environnement par défaut
  
  outputs:
    # ═══════════════════════════════════════════════════════════
    # Environnement DEV (développement local)
    # ═══════════════════════════════════════════════════════════
    dev:
      type: duckdb               # Adapter DuckDB (léger, local)
      path: /tmp/dev.duckdb
      threads: 4
    
    # ═══════════════════════════════════════════════════════════
    # Environnement PROD (Spark/Databricks)
    # ═══════════════════════════════════════════════════════════
    prod:
      type: spark
      method: thrift             # ou "odbc", "http"
      host: spark-thrift-server.company.com
      port: 10001
      schema: analytics
      threads: 8
    
    # ═══════════════════════════════════════════════════════════
    # Exemple Snowflake
    # ═══════════════════════════════════════════════════════════
    snowflake_prod:
      type: snowflake
      account: xy12345.us-east-1
      user: "{{ env_var('SNOWFLAKE_USER') }}"      # Variable d'environnement
      password: "{{ env_var('SNOWFLAKE_PASSWORD') }}"
      role: TRANSFORMER
      database: ANALYTICS
      warehouse: TRANSFORM_WH
      schema: GOLD
      threads: 8
'''

print(profiles_yml)
print("\n💡 Utilise env_var() pour les secrets, jamais en dur !")

# Exemple de model simple

model_simple = '''
-- models/staging/stg_orders.sql

-- Configuration du model (optionnel, peut être dans dbt_project.yml)
{{ config(
    materialized='view',
    schema='staging'
) }}

-- Le SELECT qui définit le model
SELECT
    order_id,
    customer_id,
    order_date,
    CAST(amount AS DECIMAL(10, 2)) AS amount,
    status,
    created_at
FROM {{ source('raw', 'orders') }}  -- Référence à une source externe
WHERE order_date >= '2024-01-01'    -- Filtrer les données récentes
'''

print(model_simple)
print("\n💡 Points clés :")
print("• {{ config(...) }} : Configuration spécifique au model")
print("• {{ source(...) }} : Référence à une table externe")
print("• Pas de CREATE TABLE : dbt le génère automatiquement")

# Exemple avec ref()

model_with_ref = '''
-- models/silver/silver_orders.sql

{{ config(materialized='table') }}

SELECT
    o.order_id,
    o.customer_id,
    c.customer_name,
    c.customer_email,
    o.order_date,
    o.amount,
    o.status
FROM {{ ref('stg_orders') }} o           -- Référence au model stg_orders
LEFT JOIN {{ ref('stg_customers') }} c   -- Référence au model stg_customers
    ON o.customer_id = c.customer_id
WHERE o.status != 'cancelled'
'''

print(model_with_ref)
print("\n📊 Ce que dbt comprend :")
print("")
print("   stg_orders ──────┐")
print("                    ├──▶ silver_orders")
print("   stg_customers ───┘")
print("")
print("dbt exécutera stg_orders et stg_customers AVANT silver_orders.")

# Définition des sources

sources_yml = '''
# models/staging/schema.yml

version: 2

sources:
  - name: raw                      # Nom logique de la source
    description: "Données brutes ingérées par Kafka/Spark"
    database: bronze               # Base de données (optionnel)
    schema: public                 # Schéma
    
    tables:
      - name: orders
        description: "Commandes brutes"
        columns:
          - name: order_id
            description: "Identifiant unique de la commande"
          - name: amount
            description: "Montant en euros"
        
        # ═══════════════════════════════════════════════════════
        # FRESHNESS : Vérifier que les données arrivent
        # ═══════════════════════════════════════════════════════
        loaded_at_field: created_at  # Colonne de timestamp
        freshness:
          warn_after: {count: 12, period: hour}   # Warning si > 12h
          error_after: {count: 24, period: hour}  # Erreur si > 24h
      
      - name: customers
        description: "Clients"
        freshness:
          warn_after: {count: 24, period: hour}
          error_after: {count: 48, period: hour}
'''

print(sources_yml)
print("\n💡 La freshness permet de détecter les problèmes d'ingestion !")
print("   Commande : dbt source freshness")

# Comparaison des matérialisations

materializations = '''
# ═══════════════════════════════════════════════════════════════
# VIEW : Pas de stockage, requête à chaque lecture
# ═══════════════════════════════════════════════════════════════
{{ config(materialized='view') }}
SELECT * FROM {{ source('raw', 'events') }}

-- dbt génère : CREATE VIEW ... AS SELECT ...
-- ✅ Toujours à jour
-- ❌ Lent si requête complexe

# ═══════════════════════════════════════════════════════════════
# TABLE : Stockage physique, reconstruction complète
# ═══════════════════════════════════════════════════════════════
{{ config(materialized='table') }}
SELECT * FROM {{ ref('stg_events') }}

-- dbt génère : CREATE TABLE ... AS SELECT ... (ou DROP + CREATE)
-- ✅ Lecture rapide
-- ❌ Rebuild complet à chaque run

# ═══════════════════════════════════════════════════════════════
# INCREMENTAL : Seulement les nouvelles données
# ═══════════════════════════════════════════════════════════════
{{ config(
    materialized='incremental',
    unique_key='event_id'           # Pour le MERGE/upsert
) }}

SELECT * FROM {{ ref('stg_events') }}
{% if is_incremental() %}           -- Condition pour les runs incrémentaux
WHERE event_time > (SELECT MAX(event_time) FROM {{ this }})
{% endif %}

-- dbt génère : INSERT INTO ... SELECT ... WHERE ...
-- ✅ Très rapide pour gros volumes
-- ❌ Plus complexe à maintenir

# ═══════════════════════════════════════════════════════════════
# EPHEMERAL : CTE, pas de table
# ═══════════════════════════════════════════════════════════════
{{ config(materialized='ephemeral') }}
SELECT DISTINCT customer_id FROM {{ ref('orders') }}

-- dbt l'injecte comme CTE dans les models qui le ref()
-- ✅ Pas de table supplémentaire
-- ❌ Recalculé à chaque utilisation
'''

print(materializations)

# Tests génériques dans schema.yml

generic_tests = '''
# models/silver/schema.yml

version: 2

models:
  - name: silver_orders
    description: "Table des commandes enrichies avec infos clients"
    
    columns:
      - name: order_id
        description: "Identifiant unique de la commande"
        tests:
          - unique              # Pas de doublons
          - not_null            # Jamais NULL
      
      - name: customer_id
        description: "Référence au client"
        tests:
          - not_null
          - relationships:      # Clé étrangère
              to: ref('stg_customers')
              field: customer_id
      
      - name: status
        description: "Statut de la commande"
        tests:
          - accepted_values:    # Valeurs autorisées
              values: ['pending', 'shipped', 'delivered', 'cancelled']
      
      - name: amount
        description: "Montant en euros"
        tests:
          - not_null
'''

print(generic_tests)
print("\n# Exécuter les tests :")
print("dbt test                    # Tous les tests")
print("dbt test --select silver_orders  # Tests d'un model")

# Tests singuliers (custom SQL)

singular_tests = '''
-- tests/assert_positive_amounts.sql

-- Ce test ÉCHOUE si la requête retourne des lignes
-- (on cherche les cas problématiques)

SELECT
    order_id,
    amount
FROM {{ ref('silver_orders') }}
WHERE amount <= 0

-- Si cette requête retourne des lignes = montants négatifs ou nuls = ÉCHEC

-- ═══════════════════════════════════════════════════════════════

-- tests/assert_orders_have_recent_dates.sql

SELECT
    order_id,
    order_date
FROM {{ ref('silver_orders') }}
WHERE order_date < '2020-01-01'
   OR order_date > CURRENT_DATE

-- Détecte les dates suspectes (trop anciennes ou dans le futur)

-- ═══════════════════════════════════════════════════════════════

-- tests/assert_revenue_consistency.sql

-- Vérifier que le total des commandes = total du dashboard
SELECT 1
FROM (
    SELECT SUM(amount) AS orders_total FROM {{ ref('silver_orders') }}
) orders
JOIN (
    SELECT SUM(revenue) AS dashboard_total FROM {{ ref('gold_dashboard') }}
) dashboard
ON 1=1
WHERE ABS(orders_total - dashboard_total) > 0.01  -- Tolérance
'''

print(singular_tests)

# Installation et utilisation des packages dbt

packages = '''
# packages.yml (à la racine du projet)

packages:
  - package: dbt-labs/dbt_utils
    version: 1.1.1
  
  - package: calogica/dbt_expectations
    version: 0.10.1
  
  - package: dbt-labs/audit_helper
    version: 0.9.0

# Installer les packages :
# dbt deps
'''

usage_examples = '''
# Utilisation dans schema.yml

models:
  - name: silver_orders
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      
      - name: amount
        tests:
          # Test dbt-expectations : valeur entre 0 et 10000
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
              max_value: 10000
          
          # Test dbt-utils : au moins 95% non-null
          - dbt_utils.not_null_proportion:
              at_least: 0.95
      
      - name: email
        tests:
          # Test dbt-expectations : format email
          - dbt_expectations.expect_column_values_to_match_regex:
              regex: "^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\\.[a-zA-Z0-9-.]+$"
'''

print("# packages.yml")
print(packages)
print("\n# Utilisation des tests avancés")
print(usage_examples)

# Documentation complète dans schema.yml

documentation = '''
# models/silver/schema.yml

version: 2

models:
  - name: silver_orders
    description: |
      Table des commandes enrichies avec les informations clients.
      
      **Source** : Kafka topic `orders` via Spark Streaming
      **Refresh** : Incrémental toutes les heures
      **Owner** : data-engineering@company.com
      
      ## Règles métier
      - Les commandes annulées sont exclues
      - Les montants sont en EUR
    
    columns:
      - name: order_id
        description: "Identifiant unique de la commande (UUID)"
        tests: [unique, not_null]
      
      - name: customer_id
        description: "Référence au client. Voir `stg_customers`."
      
      - name: amount
        description: |
          Montant total de la commande en EUR.
          Inclut les taxes, exclut les frais de livraison.
      
      - name: order_date
        description: "Date de création de la commande (timezone UTC)"
'''

print(documentation)
print("\n# Générer et servir la documentation :")
print("dbt docs generate    # Génère le site dans target/")
print("dbt docs serve       # Lance un serveur local (http://localhost:8080)")

# Installation et setup de Great Expectations

ge_setup = '''
# Installation
pip install great-expectations

# Initialiser un projet GE
great_expectations init

# Structure créée :
great_expectations/
├── great_expectations.yml      # Configuration principale
├── expectations/               # Suites d'expectations
├── checkpoints/                # Définitions des checkpoints
├── plugins/                    # Extensions custom
└── uncommitted/
    └── data_docs/              # Rapports HTML générés
'''

print(ge_setup)

# Utilisation de Great Expectations avec Python

ge_example = '''
import great_expectations as gx
from great_expectations.core.expectation_configuration import ExpectationConfiguration

# ═══════════════════════════════════════════════════════════════
# 1. Créer un contexte GE
# ═══════════════════════════════════════════════════════════════

context = gx.get_context()

# ═══════════════════════════════════════════════════════════════
# 2. Connecter une datasource (Pandas, Spark, SQL)
# ═══════════════════════════════════════════════════════════════

# Exemple avec Pandas
datasource = context.sources.add_pandas(name="my_pandas_ds")
data_asset = datasource.add_dataframe_asset(name="orders_df")

# Charger les données
import pandas as pd
df = pd.read_parquet("s3://gold/orders/")
batch_request = data_asset.build_batch_request(dataframe=df)

# ═══════════════════════════════════════════════════════════════
# 3. Créer une Expectation Suite
# ═══════════════════════════════════════════════════════════════

suite = context.add_expectation_suite(expectation_suite_name="orders_suite")

# Ajouter des expectations
suite.add_expectation(
    ExpectationConfiguration(
        expectation_type="expect_column_values_to_not_be_null",
        kwargs={"column": "order_id"}
    )
)

suite.add_expectation(
    ExpectationConfiguration(
        expectation_type="expect_column_values_to_be_between",
        kwargs={
            "column": "amount",
            "min_value": 0,
            "max_value": 50000,
            "mostly": 0.99  # 99% des valeurs doivent être dans cette plage
        }
    )
)

suite.add_expectation(
    ExpectationConfiguration(
        expectation_type="expect_column_values_to_match_regex",
        kwargs={
            "column": "email",
            "regex": r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\\.[a-zA-Z0-9-.]+$"
        }
    )
)

context.save_expectation_suite(suite)

# ═══════════════════════════════════════════════════════════════
# 4. Créer et exécuter un Checkpoint
# ═══════════════════════════════════════════════════════════════

checkpoint = context.add_or_update_checkpoint(
    name="orders_checkpoint",
    validations=[
        {
            "batch_request": batch_request,
            "expectation_suite_name": "orders_suite"
        }
    ]
)

# Exécuter la validation
result = checkpoint.run()

print(f"Validation réussie : {result.success}")

# ═══════════════════════════════════════════════════════════════
# 5. Générer les Data Docs (rapport HTML)
# ═══════════════════════════════════════════════════════════════

context.build_data_docs()
context.open_data_docs()  # Ouvre dans le navigateur
'''

print(ge_example)

# Profiling automatique avec Great Expectations

profiling = '''
from great_expectations.profile.user_configurable_profiler import UserConfigurableProfiler

# Créer un profiler
profiler = UserConfigurableProfiler(
    profile_dataset=validator,  # Validator avec les données
    excluded_expectations=None,
    ignored_columns=[],
    not_null_only=False,
    primary_or_compound_key=["order_id"],
    semantic_types_dict=None,
    table_expectations_only=False,
    value_set_threshold="many"  # "many", "few", ou un nombre
)

# Générer la suite d'expectations basée sur les données actuelles
suite = profiler.build_suite()

# Afficher les expectations générées
print(f"Nombre d'expectations générées : {len(suite.expectations)}")
for exp in suite.expectations[:5]:
    print(f"  - {exp.expectation_type}")

# Sauvegarder
context.save_expectation_suite(suite, expectation_suite_name="orders_profiled")
'''

print(profiling)
print("\n💡 Le profiling auto-génère des expectations comme :")
print("• expect_column_values_to_be_in_set (pour les catégories)")
print("• expect_column_mean_to_be_between (pour les numériques)")
print("• expect_column_proportion_of_unique_values_to_be_between")

# Configuration dbt-spark

spark_config = '''
# ~/.dbt/profiles.yml pour Spark

ecommerce_analytics:
  target: dev
  outputs:
    
    # ═══════════════════════════════════════════════════════════
    # Databricks
    # ═══════════════════════════════════════════════════════════
    databricks:
      type: databricks
      catalog: main              # Unity Catalog
      schema: analytics
      host: adb-xxx.azuredatabricks.net
      http_path: /sql/1.0/warehouses/xxx
      token: "{{ env_var('DATABRICKS_TOKEN') }}"
      threads: 4
    
    # ═══════════════════════════════════════════════════════════
    # Spark Thrift Server (EMR, local)
    # ═══════════════════════════════════════════════════════════
    spark_thrift:
      type: spark
      method: thrift
      host: spark-thrift.company.com
      port: 10001
      user: dbt_user
      schema: analytics
      threads: 4
    
    # ═══════════════════════════════════════════════════════════
    # Spark local (pour dev)
    # ═══════════════════════════════════════════════════════════
    spark_local:
      type: spark
      method: session
      schema: default
      threads: 1
'''

print(spark_config)

# Model incrémental avec Delta Lake

incremental_model = '''
-- models/gold/gold_daily_revenue.sql

{{ config(
    materialized='incremental',
    unique_key='date_customer_key',       -- Pour le MERGE
    incremental_strategy='merge',         -- MERGE INTO (vs append, delete+insert)
    file_format='delta',                  -- Format Delta Lake
    partition_by=['order_date']           -- Partitionnement
) }}

WITH source_data AS (
    SELECT
        order_date,
        customer_id,
        SUM(amount) AS daily_revenue,
        COUNT(*) AS order_count,
        -- Clé unique pour le MERGE
        CONCAT(order_date, '-', customer_id) AS date_customer_key
    FROM {{ ref('silver_orders') }}
    
    -- ═══════════════════════════════════════════════════════════
    -- FILTRE INCRÉMENTAL : Seulement les nouvelles données
    -- ═══════════════════════════════════════════════════════════
    {% if is_incremental() %}
    WHERE order_date > (
        SELECT MAX(order_date) - INTERVAL 1 DAY  -- Marge de sécurité
        FROM {{ this }}
    )
    {% endif %}
    
    GROUP BY order_date, customer_id
)

SELECT * FROM source_data
'''

print(incremental_model)
print("\n💡 Explication :")
print("• is_incremental() : True si la table existe et ce n'est pas un --full-refresh")
print("• {{ this }} : Référence à la table actuelle")
print("• INTERVAL 1 DAY : Marge pour les données en retard (late data)")
print("")
print("# Commandes :")
print("dbt run --select gold_daily_revenue          # Run incrémental")
print("dbt run --select gold_daily_revenue --full-refresh  # Rebuild complet")

# Macros Jinja réutilisables

macros = '''
-- macros/deduplicate.sql

{% macro deduplicate(relation, partition_by, order_by) %}
    {#
        Déduplique une table en gardant la ligne la plus récente.
        
        Args:
            relation: La table source
            partition_by: Colonne(s) pour identifier les doublons
            order_by: Colonne pour déterminer quelle ligne garder
    #}
    
    SELECT * FROM (
        SELECT
            *,
            ROW_NUMBER() OVER (
                PARTITION BY {{ partition_by }}
                ORDER BY {{ order_by }} DESC
            ) AS _row_num
        FROM {{ relation }}
    )
    WHERE _row_num = 1
{% endmacro %}

-- ═══════════════════════════════════════════════════════════════

-- macros/generate_surrogate_key.sql

{% macro generate_surrogate_key(columns) %}
    {#
        Génère une clé surrogate à partir de plusieurs colonnes.
    #}
    
    MD5(CONCAT(
        {% for col in columns %}
            COALESCE(CAST({{ col }} AS STRING), '_null_')
            {% if not loop.last %}, '-', {% endif %}
        {% endfor %}
    ))
{% endmacro %}

-- ═══════════════════════════════════════════════════════════════

-- macros/cents_to_euros.sql

{% macro cents_to_euros(column_name) %}
    ROUND({{ column_name }} / 100.0, 2)
{% endmacro %}
'''

usage = '''
-- Utilisation dans un model :

-- models/silver/silver_orders.sql

{{ config(materialized='table') }}

WITH deduplicated AS (
    {{ deduplicate(
        relation=ref('stg_orders'),
        partition_by='order_id',
        order_by='updated_at'
    ) }}
)

SELECT
    {{ generate_surrogate_key(['order_id', 'customer_id']) }} AS order_sk,
    order_id,
    customer_id,
    {{ cents_to_euros('amount_cents') }} AS amount,
    order_date
FROM deduplicated
'''

print("# Macros")
print(macros)
print("\n# Utilisation")
print(usage)

# GitHub Actions pour dbt CI/CD

github_actions = '''
# .github/workflows/dbt-ci.yml

name: dbt CI/CD

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  # ═══════════════════════════════════════════════════════════════
  # CI : Tests sur Pull Request
  # ═══════════════════════════════════════════════════════════════
  dbt-test:
    runs-on: ubuntu-latest
    if: github.event_name == 'pull_request'
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      
      - name: Install dbt
        run: pip install dbt-core dbt-snowflake
      
      - name: Setup dbt profile
        run: |
          mkdir -p ~/.dbt
          echo "${{ secrets.DBT_PROFILES }}" > ~/.dbt/profiles.yml
      
      - name: dbt deps
        run: dbt deps
      
      - name: dbt build (run + test)
        run: dbt build --target ci
        env:
          SNOWFLAKE_ACCOUNT: ${{ secrets.SNOWFLAKE_ACCOUNT }}
          SNOWFLAKE_USER: ${{ secrets.SNOWFLAKE_USER }}
          SNOWFLAKE_PASSWORD: ${{ secrets.SNOWFLAKE_PASSWORD }}

  # ═══════════════════════════════════════════════════════════════
  # CD : Déploiement sur main
  # ═══════════════════════════════════════════════════════════════
  dbt-deploy:
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main' && github.event_name == 'push'
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      
      - name: Install dbt
        run: pip install dbt-core dbt-snowflake
      
      - name: Setup dbt profile
        run: |
          mkdir -p ~/.dbt
          echo "${{ secrets.DBT_PROFILES }}" > ~/.dbt/profiles.yml
      
      - name: dbt deps
        run: dbt deps
      
      - name: dbt run (production)
        run: dbt run --target prod
      
      - name: dbt test (production)
        run: dbt test --target prod
'''

print(github_actions)

# DAG Airflow pour dbt

airflow_dag = '''
# dags/dbt_daily_run.py

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago

# Configuration
DBT_PROJECT_DIR = "/opt/dbt/ecommerce_analytics"
DBT_PROFILES_DIR = "/opt/dbt/profiles"

default_args = {
    "owner": "data-engineering",
    "depends_on_past": False,
    "retries": 1,
}

with DAG(
    dag_id="dbt_daily_run",
    default_args=default_args,
    description="Exécution quotidienne de dbt",
    schedule_interval="0 6 * * *",  # Tous les jours à 6h
    start_date=days_ago(1),
    catchup=False,
    tags=["dbt", "analytics"],
) as dag:
    
    # Vérifier la fraîcheur des sources
    dbt_source_freshness = BashOperator(
        task_id="dbt_source_freshness",
        bash_command=f"cd {DBT_PROJECT_DIR} && dbt source freshness --profiles-dir {DBT_PROFILES_DIR}"
    )
    
    # Exécuter les models
    dbt_run = BashOperator(
        task_id="dbt_run",
        bash_command=f"cd {DBT_PROJECT_DIR} && dbt run --profiles-dir {DBT_PROFILES_DIR}"
    )
    
    # Exécuter les tests
    dbt_test = BashOperator(
        task_id="dbt_test",
        bash_command=f"cd {DBT_PROJECT_DIR} && dbt test --profiles-dir {DBT_PROFILES_DIR}"
    )
    
    # Générer la documentation
    dbt_docs = BashOperator(
        task_id="dbt_docs_generate",
        bash_command=f"cd {DBT_PROJECT_DIR} && dbt docs generate --profiles-dir {DBT_PROFILES_DIR}"
    )
    
    # Ordre d'exécution
    dbt_source_freshness >> dbt_run >> dbt_test >> dbt_docs
'''

print(airflow_dag)

# Structure complète du mini-projet

project_structure = '''
ecommerce_dbt/
├── dbt_project.yml
├── packages.yml
│
├── models/
│   ├── staging/
│   │   ├── _sources.yml         # Sources raw
│   │   ├── stg_orders.sql
│   │   └── stg_customers.sql
│   │
│   ├── silver/
│   │   ├── _silver__models.yml  # Tests + docs
│   │   └── silver_orders.sql
│   │
│   └── gold/
│       ├── _gold__models.yml    # Tests + docs
│       ├── gold_daily_revenue.sql      # Incrémental
│       └── gold_customer_summary.sql   # Table
│
├── macros/
│   └── deduplicate.sql
│
├── tests/
│   ├── assert_positive_revenue.sql
│   └── assert_no_future_dates.sql
│
└── great_expectations/
    └── expectations/
        └── gold_orders_suite.json
'''

print(project_structure)

# Models du mini-projet

models = '''
-- ═══════════════════════════════════════════════════════════════
-- models/gold/gold_daily_revenue.sql
-- ═══════════════════════════════════════════════════════════════

{{ config(
    materialized='incremental',
    unique_key='revenue_date',
    incremental_strategy='merge'
) }}

SELECT
    order_date AS revenue_date,
    COUNT(DISTINCT order_id) AS total_orders,
    COUNT(DISTINCT customer_id) AS unique_customers,
    SUM(amount) AS total_revenue,
    AVG(amount) AS avg_order_value,
    CURRENT_TIMESTAMP() AS updated_at
FROM {{ ref('silver_orders') }}
{% if is_incremental() %}
WHERE order_date >= (SELECT MAX(revenue_date) - INTERVAL 2 DAY FROM {{ this }})
{% endif %}
GROUP BY order_date

-- ═══════════════════════════════════════════════════════════════
-- models/gold/gold_customer_summary.sql
-- ═══════════════════════════════════════════════════════════════

{{ config(materialized='table') }}

SELECT
    customer_id,
    customer_name,
    email,
    COUNT(DISTINCT order_id) AS lifetime_orders,
    SUM(amount) AS lifetime_revenue,
    AVG(amount) AS avg_order_value,
    MIN(order_date) AS first_order_date,
    MAX(order_date) AS last_order_date,
    DATEDIFF(day, MIN(order_date), MAX(order_date)) AS customer_tenure_days
FROM {{ ref('silver_orders') }}
GROUP BY customer_id, customer_name, email
'''

print(models)