02 — Pipelines et orchestration#

# 1. Gestionnaire de contexte (with)
with DAG(
    dag_id="my_dag_name",
    start_date=datetime.datetime(2021, 1, 1),
    schedule="@daily",
):
    EmptyOperator(task_id="task")

# 2. Constructeur standard
my_dag = DAG(
    dag_id="my_dag_name",
    start_date=datetime.datetime(2021, 1, 1),
    schedule="@daily",
)
EmptyOperator(task_id="task", dag=my_dag)

# 3. Décorateur @dag
@dag(start_date=datetime.datetime(2021, 1, 1), schedule="@daily")
def generate_dag():
    EmptyOperator(task_id="task")

generate_dag()

first_task >> [second_task, third_task]

@task
def get_ip():
    return my_ip_service.get_main_ip()

@task(multiple_outputs=True)
def compose_email(external_ip):
    return {
        'subject': f'Server connected from {external_ip}',
        'body': f'Your server executing Airflow is connected from the external IP {external_ip}<br>'
    }

email_info = compose_email(get_ip())

Écris-moi un DAG Airflow qui charge les commandes tous les jours.

Écris un DAG Airflow 3.

Contexte :
- Source : API REST paginée https://api.interne/commandes, auth par token dans
  la variable d'environnement API_TOKEN.
- Cible : table BigQuery analytics_raw.commandes, partitionnée sur date_commande.
- Fréquence : quotidienne à 4 h UTC.
- Volumétrie : environ 50 000 commandes par jour.

Contraintes :
- Style : TaskFlow API avec le décorateur @task.
- dag_id : "ingestion_commandes_daily".
- default_args avec retries=3.
- Via XCom, ne fais transiter que des chemins de fichiers, jamais des données.
- Le chargement doit être idempotent : relancer la même date doit donner
  le même résultat, pas des doublons.

Produis : le code, puis une liste <points_a_valider> de tout ce que tu as
supposé et que je dois confirmer.

import logging
import time

import requests

LOG = logging.getLogger(__name__)

def lire_page(url: str, token: str, page: int, tentatives: int = 3) -> dict:
    """Lit une page de l'API. Réessaie uniquement sur erreur réseau ou 5xx."""
    for essai in range(1, tentatives + 1):
        try:
            reponse = requests.get(
                url,
                headers={"Authorization": f"Bearer {token}"},
                params={"page": page},
                timeout=30,
            )
            if reponse.status_code >= 500:
                raise requests.HTTPError(f"HTTP {reponse.status_code}")
            reponse.raise_for_status()
            return reponse.json()
        except (requests.ConnectionError, requests.Timeout, requests.HTTPError) as err:
            if essai == tentatives:
                raise
            attente = 2 ** essai
            LOG.warning("Page %s: echec %s, nouvel essai dans %ss", page, err, attente)
            time.sleep(attente)
    raise RuntimeError("inatteignable")

J'ai cette erreur Airflow : KeyError: 'montant_ttc'. Pourquoi ?

<log>
[COLLE LE LOG COMPLET DE LA TÂCHE]
</log>

<code_tache>
[COLLE LE CODE DE LA FONCTION QUI A ÉCHOUÉ]
</code_tache>

<contexte>
- DAG : ingestion_commandes_daily, tâche : transform_commandes.
- Ça tournait hier sans erreur.
- Changement depuis : la source a renommé "montant" en "montant_ttc".
</contexte>

Donne-moi, dans cet ordre :
1. <cause_probable> : la cause la plus probable, et pourquoi tu le penses,
   en citant la ligne exacte du log qui l'appuie.
2. <autres_pistes> : 2 autres hypothèses, classées par probabilité.
3. <correctif> : le patch minimal.
4. <prevention> : le test ou le contrôle qui aurait attrapé ça avant la prod.

Ne suppose rien qui ne soit pas dans le log ou le code fournis.

Tu es ingénieur data senior, expert Apache Airflow.

Contexte du pipeline :
- Source : [TYPE_SOURCE_API_FICHIER_BASE] — [DETAIL_URL_OU_CHEMIN]
- Authentification : [METHODE_ET_VARIABLE_ENV]
- Cible : [TABLE_CIBLE] sur [ENTREPOT], partitionnée sur [COLONNE_PARTITION]
- Fréquence : [FREQUENCE] à [HEURE] [FUSEAU]
- Volumétrie estimée : [NB_LIGNES_PAR_EXECUTION]
- Fenêtre de traitement acceptable : [DUREE_MAX]

Contraintes techniques :
- Version d'Airflow : [VERSION]
- Style de déclaration : [TaskFlow @task | context manager with DAG | constructeur]
- dag_id : [NOM_DU_DAG]
- default_args : retries=[NB], et [AUTRES_DEFAULT_ARGS]
- XCom : uniquement des métadonnées légères (chemins, dates, compteurs).
- Le chargement doit être idempotent : rejouer une date ne crée pas de doublons.

Produis :
1. <dag> : le code complet.
2. <points_a_valider> : tout ce que tu as supposé et que je dois confirmer.
3. <risques> : les 3 façons les plus probables dont ce DAG cassera en production.

Si une information te manque, demande-la au lieu de l'inventer.

Écris une fonction Python d'ingestion.

Source : [DESCRIPTION_SOURCE]
Format de réponse : [JSON_CSV_PARQUET] — exemple d'une ligne :
[COLLE_UNE_LIGNE_REELLE_ANONYMISEE]
Cible : [TABLE_OU_FICHIER_CIBLE]

Exigences non négociables :
- Pagination : [DECRIS_LA_PAGINATION]
- Réessai avec attente croissante UNIQUEMENT sur erreurs réseau et 5xx.
  Aucun réessai sur 4xx : lève l'erreur immédiatement.
- Écriture idempotente : [DECRIS_LA_STRATEGIE_EX_SUPPRIMER_LA_PARTITION_PUIS_REECRIRE]
- Journalisation : nombre de lignes lues, nombre de lignes écrites, durée.
- Contrôle final : si lignes_ecrites != lignes_lues, lève une exception explicite.
- Lignes invalides : envoie-les dans [TABLE_DE_REJETS] au lieu de faire tomber le job.

Type-hint toutes les signatures. Pas de dépendance exotique :
utilise seulement [LISTE_DES_LIBRAIRIES_AUTORISEES].

<log>
[COLLE_LE_LOG_COMPLET]
</log>

<code_tache>
[COLLE_LE_CODE_DE_LA_TACHE_EN_ECHEC]
</code_tache>

<contexte>
- DAG : [NOM_DAG] / tâche : [NOM_TACHE]
- Dernière exécution réussie : [DATE]
- Changements depuis : [CE_QUI_A_CHANGE_CODE_SCHEMA_VOLUME_DROITS]
- Le problème est-il reproductible : [OUI_NON_PARFOIS]
</contexte>

Rends-moi :
1. <cause_probable> : la cause la plus probable, en citant la ligne exacte du log.
2. <autres_pistes> : 2 hypothèses alternatives classées par probabilité.
3. <correctif> : le patch minimal, sans refactorisation.
4. <prevention> : le test ou le contrôle qui aurait détecté ça avant la prod.

N'affirme rien qui ne soit pas visible dans le log ou le code fournis.
Si le log est insuffisant, dis-moi quelle information supplémentaire aller chercher.

<code_dag>
[COLLE_LE_CODE_COMPLET_DU_DAG]
</code_dag>

Écris la fiche de ce pipeline, en français, en six blocs :
1. Objectif métier (3 phrases max, sans jargon).
2. Sources lues : table/API, granularité, fréquence de rafraîchissement.
3. Cibles écrites : table, mode d'écriture (ajout ou remplacement), partitionnement.
4. Planification : fréquence, heure, fuseau, fenêtre de rattrapage.
5. Graphe des tâches : un tableau | tâche | dépend de | ce qu'elle fait |.
6. Points de fragilité : où ça casse le plus probablement, et pourquoi.

N'invente aucune dépendance qui ne soit pas visible dans le code. Si une
dépendance semble implicite (un fichier déposé par un autre job), signale-la
comme "dépendance implicite à confirmer".