-- Interroger une table telle qu'elle etait il y a 30 minutes
SELECT * FROM orders
AT(OFFSET => -60*30);

-- Interroger a un timestamp precis
SELECT * FROM orders
AT(TIMESTAMP => '2025-12-01 10:00:00'::TIMESTAMP);

-- Restaurer une table supprimee par erreur
UNDROP TABLE orders;

-- Creer une table a partir d'un etat passe
CREATE TABLE orders_backup AS
SELECT * FROM orders
AT(TIMESTAMP => '2025-12-01 08:00:00'::TIMESTAMP);

-- Retention configurable (0 a 90 jours selon l'edition)
ALTER TABLE orders SET DATA_RETENTION_TIME_IN_DAYS = 30;

-- Cloner une base entiere pour un environnement de dev
CREATE DATABASE dev_analytics CLONE prod_analytics;

-- Cloner un schema specifique
CREATE SCHEMA staging.test_schema CLONE staging.prod_schema;

-- Cloner une table pour des tests
CREATE TABLE orders_test CLONE orders;

-- Verifier l'espace reel consomme (quasi-zero au debut)
SELECT TABLE_NAME, ACTIVE_BYTES, TIME_TRAVEL_BYTES, RETAINED_FOR_CLONE_BYTES
FROM INFORMATION_SCHEMA.TABLE_STORAGE_METRICS
WHERE TABLE_NAME IN ('ORDERS', 'ORDERS_TEST');

-- Creer un share pour un partenaire
CREATE SHARE partner_analytics;

-- Ajouter des objets au share (vues securisees recommandees)
CREATE SECURE VIEW analytics.public.v_partner_metrics AS
SELECT region, product_category, SUM(revenue) as total_revenue,
       COUNT(DISTINCT customer_id) as unique_customers
FROM analytics.public.sales
GROUP BY region, product_category;

GRANT USAGE ON DATABASE analytics TO SHARE partner_analytics;
GRANT USAGE ON SCHEMA analytics.public TO SHARE partner_analytics;
GRANT SELECT ON VIEW analytics.public.v_partner_metrics TO SHARE partner_analytics;

-- Ajouter le compte consommateur
ALTER SHARE partner_analytics ADD ACCOUNTS = 'PARTNER_ACCT_123';

-- Configuration optimale des warehouses
CREATE WAREHOUSE wh_etl WITH
  WAREHOUSE_SIZE = 'LARGE'
  AUTO_SUSPEND = 60          -- 1 minute sans activite
  AUTO_RESUME = TRUE
  MIN_CLUSTER_COUNT = 1
  MAX_CLUSTER_COUNT = 3      -- Multi-cluster auto-scaling
  SCALING_POLICY = 'ECONOMY';

-- Resource monitor pour controler les couts
CREATE RESOURCE MONITOR monthly_budget WITH
  CREDIT_QUOTA = 5000
  FREQUENCY = MONTHLY
  START_TIMESTAMP = IMMEDIATELY
  TRIGGERS
    ON 75 PERCENT DO NOTIFY
    ON 90 PERCENT DO NOTIFY
    ON 100 PERCENT DO SUSPEND;

-- Table partitionnee par date + clusterisee
CREATE TABLE `project.dataset.events`
(
  event_id STRING,
  event_timestamp TIMESTAMP,
  user_id STRING,
  event_type STRING,
  country STRING,
  device_type STRING,
  revenue NUMERIC
)
PARTITION BY DATE(event_timestamp)       -- Partitionnement par jour
CLUSTER BY country, event_type, device_type  -- Clustering (max 4 colonnes)
OPTIONS (
  partition_expiration_days = 365,        -- Supprime les partitions > 1 an
  require_partition_filter = TRUE         -- OBLIGE a filtrer par date
);

-- Requete optimisee : scanne seulement la partition du jour + cluster prune
SELECT country, event_type, SUM(revenue) as total_revenue
FROM `project.dataset.events`
WHERE DATE(event_timestamp) = '2025-12-01'  -- Partition pruning
  AND country = 'FR'                         -- Cluster pruning
GROUP BY country, event_type;

-- Verifier les octets scannes (estimation)
-- Utiliser l'option --dry_run dans bq CLI ou l'estimation dans la console

-- Creer une reservation BI Engine (via SQL admin)
-- Reserve 10 Go de memoire pour le projet
CREATE BI_ENGINE_RESERVATION
  PROJECT_ID = 'my-analytics-project',
  LOCATION = 'EU',
  SIZE_GB = 10;

-- Les tables materialisees sont automatiquement candidates pour BI Engine
CREATE MATERIALIZED VIEW `project.dataset.mv_daily_sales` AS
SELECT
  DATE(order_timestamp) as order_date,
  product_category,
  region,
  COUNT(*) as order_count,
  SUM(amount) as total_revenue,
  AVG(amount) as avg_order_value
FROM `project.dataset.orders`
GROUP BY 1, 2, 3;

-- BigQuery ML : creer un modele de prediction directement en SQL
CREATE OR REPLACE MODEL `project.dataset.churn_model`
OPTIONS (
  model_type = 'LOGISTIC_REG',
  input_label_cols = ['churned'],
  auto_class_weights = TRUE,
  data_split_method = 'AUTO_SPLIT'
) AS
SELECT
  days_since_last_purchase,
  total_orders,
  avg_order_value,
  support_tickets_count,
  email_open_rate,
  churned
FROM `project.dataset.customer_features`
WHERE signup_date < '2025-01-01';

-- Predire le churn sur les nouveaux clients
SELECT customer_id, predicted_churned, predicted_churned_probs
FROM ML.PREDICT(MODEL `project.dataset.churn_model`,
  (SELECT * FROM `project.dataset.customer_features`
   WHERE signup_date >= '2025-01-01'))
ORDER BY predicted_churned_probs[OFFSET(1)].prob DESC
LIMIT 100;

# Creer une table Delta avec Spark
from delta.tables import DeltaTable

# Ecriture initiale
df_events = spark.read.json("s3://raw-data/events/2025/12/")
df_events.write.format("delta") \
    .partitionBy("event_date") \
    .mode("overwrite") \
    .save("s3://lakehouse/bronze/events")

# MERGE (upsert) - operation impossible avec un Data Lake classique
delta_table = DeltaTable.forPath(spark, "s3://lakehouse/silver/customers")
delta_table.alias("target").merge(
    df_new_customers.alias("source"),
    "target.customer_id = source.customer_id"
).whenMatchedUpdate(set={
    "email": "source.email",
    "last_active": "source.last_active",
    "updated_at": "current_timestamp()"
}).whenNotMatchedInsert(values={
    "customer_id": "source.customer_id",
    "email": "source.email",
    "last_active": "source.last_active",
    "created_at": "current_timestamp()",
    "updated_at": "current_timestamp()"
}).execute()

-- Time Travel avec Delta Lake
-- Voir l'historique des modifications
DESCRIBE HISTORY lakehouse.silver.customers;

-- Requeter une version precedente
SELECT * FROM lakehouse.silver.customers VERSION AS OF 42;

-- Requeter a un timestamp
SELECT * FROM lakehouse.silver.customers TIMESTAMP AS OF '2025-12-01 10:00:00';

-- Restaurer une version precedente
RESTORE TABLE lakehouse.silver.customers TO VERSION AS OF 42;

-- Schema Evolution : ajouter une colonne automatiquement
ALTER TABLE lakehouse.silver.customers
SET TBLPROPERTIES ('delta.columnMapping.mode' = 'name');

-- OPTIMIZE : compacter les petits fichiers (small file problem)
OPTIMIZE lakehouse.silver.customers
ZORDER BY (country, signup_date);

-- VACUUM : supprimer les fichiers obsoletes (liberer du stockage)
VACUUM lakehouse.silver.customers RETAIN 168 HOURS;

-- Strategie : utiliser dbt comme couche d'abstraction
-- Le meme modele dbt peut cibler Snowflake, BigQuery ou Databricks
-- en changeant simplement le profil de connexion

-- models/marts/finance/revenue_monthly.sql
-- Ce modele fonctionne sur les 3 plateformes avec dbt
{{ config(
    materialized='table',
    partition_by={'field': 'month_date', 'data_type': 'date'}  -- BigQuery
    -- cluster_by=['region', 'product_line']  -- Snowflake/Databricks
) }}

SELECT
    DATE_TRUNC('month', order_date) AS month_date,  -- ANSI SQL
    region,
    product_line,
    COUNT(DISTINCT customer_id) AS unique_customers,
    SUM(amount) AS total_revenue,
    AVG(amount) AS avg_order_value
FROM {{ ref('stg_orders') }}
WHERE order_status = 'completed'
GROUP BY 1, 2, 3

# dbt_project.yml - Configuration du projet
name: 'analytics'
version: '1.0.0'
config-version: 2

profile: 'analytics'  # Connexion au DWH (Snowflake, BigQuery, etc.)

model-paths: ["models"]
test-paths: ["tests"]
macro-paths: ["macros"]
seed-paths: ["seeds"]

models:
  analytics:
    staging:
      +materialized: view        # Les staging models sont des vues
      +schema: staging
    intermediate:
      +materialized: ephemeral   # Pas materialise, inline dans les requetes
    marts:
      +materialized: table       # Les marts sont des tables physiques
      +schema: analytics

# models/staging/sources.yml - Declaration des sources
version: 2

sources:
  - name: raw_ecommerce
    database: raw_database
    schema: ecommerce
    description: "Donnees brutes extraites par Fivetran depuis Shopify"
    tables:
      - name: orders
        description: "Table des commandes Shopify"
        loaded_at_field: _fivetran_synced
        freshness:
          warn_after: {count: 12, period: hour}
          error_after: {count: 24, period: hour}
        columns:
          - name: order_id
            description: "Identifiant unique de la commande"
            tests:
              - unique
              - not_null
      - name: customers
        description: "Table des clients Shopify"
      - name: products
        description: "Catalogue produits"

-- models/staging/stg_orders.sql
-- Staging : nettoyage, renommage, typage. 1 modele par source.
WITH source AS (
    SELECT * FROM {{ source('raw_ecommerce', 'orders') }}
),

renamed AS (
    SELECT
        id AS order_id,
        customer_id,
        CAST(created_at AS TIMESTAMP) AS order_timestamp,
        DATE(created_at) AS order_date,
        status AS order_status,
        CAST(total_price AS NUMERIC) AS order_amount,
        currency,
        LOWER(TRIM(shipping_country)) AS shipping_country,
        _fivetran_synced AS loaded_at
    FROM source
    WHERE id IS NOT NULL  -- Exclure les lignes corrompues
)

SELECT * FROM renamed

-- models/intermediate/int_orders_enriched.sql
-- Intermediate : jointures, logique metier, enrichissement
WITH orders AS (
    SELECT * FROM {{ ref('stg_orders') }}
),

customers AS (
    SELECT * FROM {{ ref('stg_customers') }}
),

products AS (
    SELECT * FROM {{ ref('stg_order_items') }}
)

SELECT
    o.order_id,
    o.order_timestamp,
    o.order_date,
    o.order_status,
    o.order_amount,
    c.customer_id,
    c.customer_segment,
    c.first_order_date,
    c.country AS customer_country,
    -- Calcul metier : est-ce une premiere commande ?
    CASE WHEN o.order_date = c.first_order_date
         THEN TRUE ELSE FALSE END AS is_first_order,
    -- Nombre de jours depuis la derniere commande
    DATEDIFF('day', c.last_order_date, o.order_date) AS days_since_last_order
FROM orders o
LEFT JOIN customers c ON o.customer_id = c.customer_id

-- models/marts/finance/fct_revenue.sql
-- Mart : modele final pret pour la BI
{{ config(
    materialized='incremental',
    unique_key='order_id',
    on_schema_change='append_new_columns'
) }}

SELECT
    order_id,
    order_date,
    customer_id,
    customer_segment,
    customer_country,
    order_amount,
    is_first_order,
    -- Metriques derivees
    SUM(order_amount) OVER (
        PARTITION BY customer_id
        ORDER BY order_date
        ROWS UNBOUNDED PRECEDING
    ) AS customer_lifetime_value
FROM {{ ref('int_orders_enriched') }}
WHERE order_status = 'completed'

{% if is_incremental() %}
    AND order_date > (SELECT MAX(order_date) FROM {{ this }})
{% endif %}

# models/marts/schema.yml
version: 2

models:
  - name: fct_revenue
    description: "Table de faits des revenus par commande completee"
    columns:
      - name: order_id
        description: "Identifiant unique de la commande"
        tests:
          - unique
          - not_null

      - name: order_date
        description: "Date de la commande"
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: "'2020-01-01'"
              max_value: "CURRENT_DATE()"

      - name: order_status
        description: "Statut de la commande"
        tests:
          - accepted_values:
              values: ['completed', 'refunded', 'partially_refunded']

      - name: customer_id
        description: "Reference vers la dimension client"
        tests:
          - not_null
          - relationships:
              to: ref('dim_customers')
              field: customer_id

      - name: order_amount
        description: "Montant total de la commande en euros"
        tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: ">= 0"

-- tests/assert_revenue_not_negative.sql
-- Un test custom passe si la requete retourne 0 ligne
-- Ici on verifie qu'aucun client n'a un revenue cumule negatif

SELECT
    customer_id,
    SUM(order_amount) AS total_revenue
FROM {{ ref('fct_revenue') }}
GROUP BY customer_id
HAVING SUM(order_amount) < 0

-- tests/assert_daily_revenue_consistency.sql
-- Verifier que le revenue journalier ne s'ecarte pas
-- de plus de 50% de la moyenne mobile sur 7 jours

WITH daily_revenue AS (
    SELECT
        order_date,
        SUM(order_amount) AS daily_total,
        AVG(SUM(order_amount)) OVER (
            ORDER BY order_date
            ROWS BETWEEN 7 PRECEDING AND 1 PRECEDING
        ) AS avg_7d
    FROM {{ ref('fct_revenue') }}
    GROUP BY order_date
)

SELECT
    order_date,
    daily_total,
    avg_7d,
    ABS(daily_total - avg_7d) / NULLIF(avg_7d, 0) AS pct_deviation
FROM daily_revenue
WHERE avg_7d IS NOT NULL
  AND ABS(daily_total - avg_7d) / NULLIF(avg_7d, 0) > 0.5
  AND order_date >= DATEADD('day', -7, CURRENT_DATE())

# packages.yml - Dependances du projet
packages:
  - package: dbt-labs/dbt_utils
    version: ">=1.1.0"

  - package: calogica/dbt_expectations
    version: ">=0.10.0"

  - package: dbt-labs/codegen
    version: ">=0.12.0"

# Installer avec : dbt deps

# models/marts/schema.yml - Documentation enrichie
version: 2

models:
  - name: fct_revenue
    description: |
      Table de faits contenant une ligne par commande completee.
      **Grain** : une ligne par order_id.
      **Source** : donnees Shopify via Fivetran.
      **Rafraichissement** : quotidien a 06h UTC.
      **Owner** : equipe Finance Data (finance-data@company.com).

      ### Logique metier
      - Seules les commandes avec statut 'completed' sont incluses
      - Le montant est en euros, apres conversion si necessaire
      - Le customer_lifetime_value est calcule cumulativement

      ### Exemples de requetes
      ```sql
      -- Revenue mensuel par segment
      SELECT customer_segment, DATE_TRUNC('month', order_date),
             SUM(order_amount) FROM fct_revenue GROUP BY 1, 2
      ```
    columns:
      - name: order_id
        description: "Cle primaire - identifiant unique Shopify de la commande"
      - name: customer_lifetime_value
        description: |
          Somme cumulee de tous les montants de commande du client,
          ordonnee par date de commande. Calcul : window function
          SUM(order_amount) OVER (PARTITION BY customer_id ORDER BY order_date)

# models/exposures.yml - Documenter les consommateurs
version: 2

exposures:
  - name: monthly_revenue_dashboard
    type: dashboard
    description: "Dashboard Looker utilise par la direction financiere"
    owner:
      name: Marie Dupont
      email: marie.dupont@company.com
    url: https://looker.company.com/dashboards/42
    depends_on:
      - ref('fct_revenue')
      - ref('dim_customers')
      - ref('dim_products')
    maturity: high

  - name: churn_prediction_model
    type: ml
    description: "Modele ML de prediction du churn client"
    owner:
      name: Ahmed Ben Ali
      email: ahmed.benali@company.com
    depends_on:
      - ref('fct_revenue')
      - ref('dim_customers')

# .github/workflows/dbt-ci.yml
name: dbt CI
on:
  pull_request:
    branches: [main]
    paths:
      - 'models/**'
      - 'tests/**'
      - 'macros/**'

jobs:
  dbt-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install dbt-snowflake
      - run: dbt deps
      - run: dbt build --select state:modified+ --defer --state ./prod-manifest
        env:
          DBT_PROFILES_DIR: ./ci_profiles
          SNOWFLAKE_ACCOUNT: ${{ secrets.SNOWFLAKE_ACCOUNT }}
          SNOWFLAKE_PASSWORD: ${{ secrets.SNOWFLAKE_PASSWORD }}

-- models/staging/stg_events.sql
{{
  config(
    materialized='incremental',
    unique_key='event_id',
    incremental_strategy='merge',
    on_schema_change='append_new_columns'
  )
}}

SELECT
    event_id,
    user_id,
    event_type,
    event_properties,
    created_at,
    _loaded_at
FROM {{ source('raw', 'events') }}

{% if is_incremental() %}
  WHERE _loaded_at > (SELECT MAX(_loaded_at) FROM {{ this }})
{% endif %}

-- models/intermediate/int_daily_revenue.sql
{{
  config(
    materialized='incremental',
    incremental_strategy='insert_overwrite',
    partition_by={
      "field": "order_date",
      "data_type": "date",
      "granularity": "day"
    },
    cluster_by=["region", "product_category"]
  )
}}

SELECT
    DATE(order_timestamp) AS order_date,
    region,
    product_category,
    COUNT(*) AS order_count,
    SUM(amount) AS total_revenue
FROM {{ ref('stg_orders') }}

{% if is_incremental() %}
  WHERE DATE(order_timestamp) >= DATE_SUB(
    (SELECT MAX(order_date) FROM {{ this }}),
    INTERVAL 3 DAY  -- Lookback pour rattraper les retards
  )
{% endif %}

GROUP BY 1, 2, 3

-- snapshots/snap_customers.sql
{% snapshot snap_customers %}
{{
  config(
    target_schema='snapshots',
    unique_key='customer_id',
    strategy='timestamp',
    updated_at='updated_at',
    invalidate_hard_deletes=True
  )
}}

SELECT
    customer_id,
    company_name,
    subscription_plan,
    mrr,
    status,
    updated_at
FROM {{ source('crm', 'customers') }}

{% endsnapshot %}

-- macros/generate_surrogate_key.sql
{% macro generate_surrogate_key(field_list) %}
    {% set fields = [] %}
    {% for field in field_list %}
        {% do fields.append(
            "COALESCE(CAST(" ~ field ~ " AS VARCHAR), '_null_')"
        ) %}
    {% endfor %}
    {{ dbt_utils.hash(fields | join(" || '|' || ")) }}
{% endmacro %}

-- macros/cents_to_dollars.sql
{% macro cents_to_dollars(column_name, precision=2) %}
    ROUND({{ column_name }} / 100.0, {{ precision }})
{% endmacro %}

-- Utilisation dans un modele :
-- SELECT {{ generate_surrogate_key(['order_id', 'line_item_id']) }} AS sk,
--        {{ cents_to_dollars('amount_cents') }} AS amount_dollars

packages:
  - package: dbt-labs/dbt_utils
    version: [">=1.0.0", "<2.0.0"]

  - package: dbt-labs/dbt_expectations
    version: [">=0.8.0", "<1.0.0"]

  - package: calogica/dbt_date
    version: [">=0.7.0", "<1.0.0"]

  - git: "https://github.com/mon-org/dbt-internal-macros.git"
    revision: v1.2.0

-- dbt_project.yml
models:
  my_project:
    marts:
      +pre-hook:
        - "SET query_tag = 'dbt_marts_{{ this.name }}'"
      +post-hook:
        - "GRANT SELECT ON {{ this }} TO ROLE analyst_role"
        - "ALTER TABLE {{ this }} SET DATA_RETENTION_TIME_IN_DAYS = 7"

-- Ou directement dans un modele :
{{
  config(
    materialized='table',
    post_hook=[
      "CREATE INDEX IF NOT EXISTS idx_{{ this.name }}_user
       ON {{ this }} (user_id)",
      "ANALYZE {{ this }}"
    ]
  )
}}

name: dbt CI

on:
  pull_request:
    branches: [main]
    paths:
      - 'models/**'
      - 'macros/**'
      - 'tests/**'
      - 'dbt_project.yml'

env:
  DBT_PROFILES_DIR: ./ci_profiles
  SNOWFLAKE_ACCOUNT: ${{ secrets.SNOWFLAKE_ACCOUNT }}
  SNOWFLAKE_USER: ${{ secrets.SNOWFLAKE_USER }}
  SNOWFLAKE_PASSWORD: ${{ secrets.SNOWFLAKE_PASSWORD }}

jobs:
  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install sqlfluff dbt-sqlfluff
      - run: sqlfluff lint models/ --dialect snowflake

  slim-ci:
    needs: lint
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0

      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - run: pip install dbt-snowflake~=1.7.0

      - name: Telecharger manifest de production
        run: |
          aws s3 cp s3://dbt-artifacts/prod/manifest.json \
            ./target-defer/manifest.json

      - name: dbt deps
        run: dbt deps

      - name: Slim CI - Build modeles modifies
        run: |
          dbt build \
            --select state:modified+ \
            --defer \
            --state ./target-defer \
            --target ci \
            --fail-fast

      - name: Commenter la PR
        if: always()
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const results = JSON.parse(
              fs.readFileSync('target/run_results.json')
            );
            const summary = results.results.map(r =>
              `| ${r.unique_id} | ${r.status} | ${r.execution_time}s |`
            ).join('\n');
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: `## Resultats dbt CI\n| Modele | Statut | Temps |\n|---|---|---|\n${summary}`
            });

# Selectionner uniquement les modeles modifies
dbt run --select state:modified --state ./prod-manifest/

# Modeles modifies + leurs enfants (downstream)
dbt run --select state:modified+ --state ./prod-manifest/

# Tester les modeles modifies et 1 niveau d'enfants
dbt test --select state:modified+1 --state ./prod-manifest/

# Combiner avec defer pour utiliser les tables prod non-modifiees
dbt build --select state:modified+ \
          --defer \
          --state ./prod-manifest/ \
          --target ci

# Verifier les changements de schema uniquement
dbt run --select state:modified --state ./prod-manifest/ \
        --vars '{"schema_check_only": true}'

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.utils.task_group import TaskGroup

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

with DAG(
    dag_id='ecommerce_daily_pipeline',
    default_args=default_args,
    description='Pipeline quotidien e-commerce ELT',
    schedule_interval='0 6 * * *',  # Tous les jours a 6h
    start_date=datetime(2024, 1, 1),
    catchup=False,
    max_active_runs=1,
    tags=['ecommerce', 'daily', 'production'],
) as dag:

    # Sensor : attendre que l'ingestion soit terminee
    wait_for_ingestion = ExternalTaskSensor(
        task_id='wait_for_ingestion',
        external_dag_id='raw_data_ingestion',
        external_task_id='load_complete',
        timeout=3600,
        poke_interval=60,
        mode='reschedule',  # Libere le worker entre les checks
    )

    # Verifier la fraicheur des donnees
    check_freshness = BashOperator(
        task_id='check_source_freshness',
        bash_command='cd /opt/dbt && dbt source freshness --select source:postgres',
        pool='dbt_pool',
    )

    # Groupe de taches : dbt build
    with TaskGroup('dbt_build', tooltip='Build dbt par couche') as dbt_group:
        staging = BashOperator(
            task_id='dbt_staging',
            bash_command='cd /opt/dbt && dbt run --select tag:staging',
            pool='dbt_pool',
        )
        intermediate = BashOperator(
            task_id='dbt_intermediate',
            bash_command='cd /opt/dbt && dbt run --select tag:intermediate',
            pool='dbt_pool',
        )
        marts = BashOperator(
            task_id='dbt_marts',
            bash_command='cd /opt/dbt && dbt run --select tag:marts',
            pool='dbt_pool',
        )
        staging >> intermediate >> marts

    # Tests dbt
    dbt_test = BashOperator(
        task_id='dbt_test',
        bash_command='cd /opt/dbt && dbt test --select tag:marts',
        pool='dbt_pool',
    )

    # Notification
    def notify_success(**context):
        execution_date = context['execution_date']
        print(f"Pipeline complete pour {execution_date}")

    notify = PythonOperator(
        task_id='notify_success',
        python_callable=notify_success,
    )

    wait_for_ingestion >> check_freshness >> dbt_group >> dbt_test >> notify

# Pousser des donnees vers XCom
def extract_stats(**context):
    stats = {'total_rows': 150000, 'new_rows': 1200, 'errors': 3}
    context['ti'].xcom_push(key='pipeline_stats', value=stats)
    return stats

# Tirer des donnees depuis XCom
def check_quality(**context):
    stats = context['ti'].xcom_pull(
        task_ids='extract_stats',
        key='pipeline_stats'
    )
    if stats['errors'] > 100:
        raise ValueError(f"Trop d'erreurs : {stats['errors']}")
    print(f"Qualite OK : {stats['new_rows']} nouvelles lignes")

import dagster as dg
import pandas as pd

# Asset source
@dg.asset(
    description="Commandes brutes depuis PostgreSQL",
    group_name="staging",
    compute_kind="sql",
    metadata={"owner": "data-team", "freshness_policy": "daily"},
)
def raw_orders(context: dg.AssetExecutionContext) -> pd.DataFrame:
    """Extraction des commandes depuis la base transactionnelle."""
    query = "SELECT * FROM orders WHERE date >= CURRENT_DATE - INTERVAL '1 day'"
    df = pd.read_sql(query, context.resources.postgres_conn)
    context.log.info(f"Extrait {len(df)} commandes")
    context.add_output_metadata({
        "row_count": len(df),
        "date_range": f"{df['ordered_at'].min()} - {df['ordered_at'].max()}"
    })
    return df


# Asset dependant
@dg.asset(
    description="Commandes nettoyees et enrichies",
    group_name="intermediate",
    compute_kind="pandas",
    deps=[raw_orders],  # Dependance explicite
)
def clean_orders(context: dg.AssetExecutionContext, raw_orders: pd.DataFrame) -> pd.DataFrame:
    """Nettoyage : suppression des doublons, typage, filtrage."""
    df = raw_orders.copy()
    df = df.drop_duplicates(subset=['order_id'])
    df['order_total'] = df['total_amount_cents'] / 100.0
    df['ordered_at'] = pd.to_datetime(df['ordered_at'])
    df = df[df['order_total'] > 0]

    context.log.info(f"Apres nettoyage : {len(df)} commandes valides")
    return df


# Asset mart
@dg.asset(
    description="Revenu quotidien agrege",
    group_name="marts",
    compute_kind="pandas",
    freshness_policy=dg.FreshnessPolicy(maximum_lag_minutes=60 * 8),
)
def daily_revenue(context: dg.AssetExecutionContext, clean_orders: pd.DataFrame) -> pd.DataFrame:
    """Agregation du revenu par jour pour le dashboard finance."""
    df = clean_orders.groupby(clean_orders['ordered_at'].dt.date).agg(
        order_count=('order_id', 'nunique'),
        gross_revenue=('order_total', 'sum'),
        avg_order_value=('order_total', 'mean'),
    ).reset_index()

    context.add_output_metadata({
        "row_count": len(df),
        "total_revenue": float(df['gross_revenue'].sum()),
    })
    return df

from dagster import op, graph, job, In, Out, String

@op(
    description="Valide la qualite des donnees",
    ins={"table_name": In(String)},
    out={"validation_result": Out(bool)},
)
def validate_data_quality(context, table_name: str) -> bool:
    """Execute des checks de qualite sur une table."""
    checks = {
        'null_check': f"SELECT COUNT(*) FROM {table_name} WHERE id IS NULL",
        'dup_check': f"SELECT COUNT(*) - COUNT(DISTINCT id) FROM {table_name}",
    }
    all_passed = True
    for name, query in checks.items():
        result = context.resources.db.execute(query)
        if result > 0:
            context.log.warning(f"Check {name} echoue : {result} problemes")
            all_passed = False
    return all_passed


@op
def send_alert(context, validation_passed: bool):
    if not validation_passed:
        context.resources.slack.send_message(
            channel="#data-alerts",
            text="Probleme de qualite detecte !"
        )


@graph
def quality_check_pipeline():
    result = validate_data_quality()
    send_alert(result)


# Transformer le graph en job executable
quality_job = quality_check_pipeline.to_job(
    resource_defs={
        "db": snowflake_resource,
        "slack": slack_resource,
    }
)

from dagster import IOManager, io_manager, InputContext, OutputContext
import pandas as pd

class SnowflakeIOManager(IOManager):
    """Stocke et charge les assets depuis Snowflake."""

    def __init__(self, connection_string: str, schema: str):
        self.connection_string = connection_string
        self.schema = schema

    def handle_output(self, context: OutputContext, obj: pd.DataFrame):
        table_name = context.asset_key.to_user_string().replace("/", "_")
        full_table = f"{self.schema}.{table_name}"
        obj.to_sql(full_table, self.connection_string, if_exists='replace')
        context.log.info(f"Ecrit {len(obj)} lignes dans {full_table}")

    def load_input(self, context: InputContext) -> pd.DataFrame:
        table_name = context.asset_key.to_user_string().replace("/", "_")
        full_table = f"{self.schema}.{table_name}"
        return pd.read_sql(f"SELECT * FROM {full_table}", self.connection_string)

@io_manager(config_schema={"connection_string": str, "schema": str})
def snowflake_io_manager(context):
    return SnowflakeIOManager(
        connection_string=context.resource_config["connection_string"],
        schema=context.resource_config["schema"],
    )

from prefect import flow, task, get_run_logger
from prefect.tasks import task_input_hash
from datetime import timedelta
import pandas as pd
import httpx

@task(
    name="extract-api-data",
    description="Extrait les donnees depuis l'API externe",
    retries=3,
    retry_delay_seconds=30,
    cache_key_fn=task_input_hash,
    cache_expiration=timedelta(hours=1),
    tags=["extract", "api"],
)
def extract_from_api(endpoint: str, date: str) -> dict:
    """Appel API avec retry automatique et cache."""
    logger = get_run_logger()
    response = httpx.get(
        f"https://api.ecommerce.com/v2/{endpoint}",
        params={"date": date},
        timeout=30,
    )
    response.raise_for_status()
    data = response.json()
    logger.info(f"Extrait {len(data['results'])} enregistrements depuis {endpoint}")
    return data


@task(
    name="transform-orders",
    tags=["transform"],
)
def transform_orders(raw_data: dict) -> pd.DataFrame:
    """Transforme les donnees brutes en DataFrame structure."""
    logger = get_run_logger()
    df = pd.DataFrame(raw_data['results'])
    df['order_total'] = df['amount_cents'] / 100.0
    df['ordered_at'] = pd.to_datetime(df['ordered_at'])
    df = df[df['order_total'] > 0]

    logger.info(f"Transforme {len(df)} commandes valides")
    return df


@task(
    name="load-to-warehouse",
    retries=2,
    retry_delay_seconds=60,
    tags=["load", "snowflake"],
)
def load_to_snowflake(df: pd.DataFrame, table_name: str):
    """Charge les donnees dans Snowflake."""
    logger = get_run_logger()
    from prefect_snowflake import SnowflakeConnector
    connector = SnowflakeConnector.load("prod-snowflake")
    with connector.get_connection() as conn:
        df.to_sql(table_name, conn, if_exists='append', index=False)
    logger.info(f"Charge {len(df)} lignes dans {table_name}")


@flow(
    name="ecommerce-daily-pipeline",
    description="Pipeline ELT quotidien e-commerce",
    version="2.1.0",
    retries=1,
    retry_delay_seconds=300,
    log_prints=True,
)
def daily_pipeline(target_date: str = None):
    """Pipeline principal : extract, transform, load."""
    from datetime import date
    target = target_date or str(date.today())

    # Extract (parallel par defaut si independant)
    orders_raw = extract_from_api("orders", target)
    products_raw = extract_from_api("products", target)

    # Transform
    orders_clean = transform_orders(orders_raw)

    # Load
    load_to_snowflake(orders_clean, "staging.orders")

    print(f"Pipeline termine pour {target}")


# Execution locale pour test
if __name__ == "__main__":
    daily_pipeline(target_date="2024-03-15")

# prefect.yaml
deployments:
  - name: ecommerce-daily
    entrypoint: pipelines/daily.py:daily_pipeline
    work_pool:
      name: production-k8s
      job_variables:
        image: registry.company.com/data-pipelines:latest
        cpu: "2000m"
        memory: "4Gi"
    schedule:
      cron: "0 6 * * *"
      timezone: "Europe/Paris"
    parameters:
      target_date: null  # Sera calcule automatiquement
    tags: ["production", "daily", "ecommerce"]

  - name: ecommerce-backfill
    entrypoint: pipelines/daily.py:daily_pipeline
    work_pool:
      name: production-k8s
      job_variables:
        image: registry.company.com/data-pipelines:latest
        cpu: "4000m"
        memory: "8Gi"
    tags: ["backfill", "ecommerce"]

# Creer un work pool Kubernetes
prefect work-pool create production-k8s --type kubernetes

# Demarrer un worker
prefect worker start --pool production-k8s

# Deployer les flows
prefect deploy --all

# Lancer manuellement un flow
prefect deployment run 'ecommerce-daily-pipeline/ecommerce-daily' \
  --param target_date=2024-03-01

# Backfill sur une plage de dates
for d in $(seq -w 1 31); do
  prefect deployment run 'ecommerce-daily-pipeline/ecommerce-backfill' \
    --param target_date=2024-03-$d
done

from prefect_snowflake import SnowflakeConnector
from prefect.blocks.notifications import SlackWebhook
from prefect.blocks.system import Secret

# Configurer un block Snowflake (via UI ou code)
snowflake_block = SnowflakeConnector(
    account="myorg-account",
    user="dbt_user",
    password="...",  # Ou reference a un Secret block
    database="analytics",
    warehouse="transform_wh",
    schema="public",
    role="transformer",
)
snowflake_block.save("prod-snowflake", overwrite=True)

# Block Slack pour les notifications
slack = SlackWebhook(url="https://hooks.slack.com/services/...")
slack.save("data-alerts", overwrite=True)

# Utilisation dans un flow
@flow
def monitored_pipeline():
    connector = SnowflakeConnector.load("prod-snowflake")
    slack = SlackWebhook.load("data-alerts")

    try:
        # ... logique pipeline
        pass
    except Exception as e:
        slack.notify(f"Pipeline echoue : {str(e)}")
        raise