ajout de titres au blocs de code

Author: Michael-cd30

content/articles/2025/2025-05-27_taradata_extract_load_mapillary.md

@@ -72,7 +72,7 @@ Plusieurs options sont également disponibles pour créer un _DAG_ mais avec _Ta
 
 Ci-dessous, un exemple de _DAG_ pour récupérer chaque heure la hauteur d'eau du Gardon à Anduze grâce à l'[API Hydrométrie de Hubeau](https://hubeau.eaufrance.fr/page/api-hydrometrie).
 
-```py
+```py title="_DAG_ d'extraction/chargement de données depuis Hubeau"
 import requests
 from airflow.decorators import dag, task
 from datetime import datetime
@@ -149,7 +149,7 @@ Après cette entrée en matière sur Apache Airflow, voyons maintenant le script
 
 La première tâche de notre _DAG_ consiste en la création du schéma d'accueil dans l'entrepôt PostgreSQL.
 
-```py
+```py title="Tâche de création du schéma PostgreSQL"
 create_schema_task = postgresql_tasks.create_schema(taradata_storage, target_schema)
 ```
 
@@ -173,7 +173,7 @@ def create_schema(pg_storage: BasePostgreSQLDataStorage, schema: str):
 
 Avec cette seconde tâche, nous créons la table d'accueil des données que nous nous apprêtons à extraire.
 
-```py
+```py title="Tâche de création de la table temporaire de chargement"
 create_table_task = postgresql_tasks.execute_sql_statement.override(task_id = "créer_table_temporaire")(
     taradata_storage,
     """
@@ -199,7 +199,7 @@ La transformation de ces données JSON en quelque chose d'exploitable, avec [QGI
 
 Avant de poursuivre, nous devons chaîner les tâches. En effet, il ne faut pas essayer de créer la table avant d'avoir terminé la création du schéma. Ceci est fait grâce à l'opérateur `>>`.
 
-```py
+```py title="Chaînage des tâches"
 create_schema_task >> create_table_task
 ```
 
@@ -253,7 +253,7 @@ repartition_aleatoire as (
 
 Ne reste plus qu'à insérer cette répartition aléatoire dans notre table de chargement, ce qui donne la requête globale suivante.
 
-```sql
+```sql title="Requête de calcul de l'emprise d'extraction et de répartition du travail"
 with emprise as (
     select ST_Collect(geom) as geom
     from troncons_wgs84
@@ -287,7 +287,7 @@ Nous pouvons alors afficher le résultat de cette répartition dans QGIS. De l'a
 
 Tout est prêt pour extraire et charger les données. Histoire de ne pas rentrer directement dans le dur, analysons d'abord l'entête et le pseudo-code de la tâche.
 
-```py
+```py title="Entête et pseudo-code de la tâche d'extraction/chargement"
 @task(task_id = "extraire_charger_features", retries = 3)
 def extract_load_features(extract_load_task_id: int):
     """
@@ -323,7 +323,7 @@ De nouvelles cellules sont donc potentiellement créées à la sortie de la bouc
 
 Cette étape consiste simplement à exécuter la requête suivante.
 
-```sql
+```sql title="Requête de récupération de la liste des cellules à extraire/charger"
 select
     geom,
     ST_XMin(geom) as x_min,
@@ -348,7 +348,7 @@ Sur chaque cellule, l'extraction se fait via un appel HTTP à l'API en passant e
 
 Nous avons encapsulé cet appel dans la fonction ci-dessous.
 
-```py
+```py title="Fonction d'appel à l'API Mapillary"
 def call_map_features_api(cell: dict):
     """
     Appel à l'API d'extraction des "features" au format JSON pour une cellule donnée.
@@ -373,7 +373,7 @@ Par ailleurs, Apache Airflow propose [la gestion de connexions](https://airflow.
 
 Le chargement des données consiste en la mise à jour du champs `ìnformations` (type `jsonb`) de la table de chargement pour la géometrie correspondante.
 
-```sql
+```sql title="Requête de chargement des données"
 update tmp_features
 set informations = %(informations)s
 where ST_Equals(geom, (%(geom)s)::geometry);
@@ -385,7 +385,7 @@ Les cellules pour lesquelles 2000 _features_ ont été retournées sont supprim
 
 La suppression et l'ajout sont réalisés en un seul ordre SQL grâce au mot-clé [`returning`](https://www.postgresql.org/docs/current/dml-returning.html) qui permet de récupérer tout ou partie des champs des lignes modifiées.
 
-```sql
+```sql title="Requête de découpage des cellules contenant plus de 2000 _features_"
 with cellules_a_diviser as (
     delete
     from tmp_features
@@ -410,7 +410,7 @@ La tâche 4 est construite de sorte à traiter une sous-partie des quelques 4000
 
 Il faut donc invoquer autant de fois que souhaité la tâche pour chacune des sous-parties. Pour cela, nous mettons ces invocations dans une liste Python.
 
-```py
+```py title="Création de N tâches d'extraction/chargement"
 extract_load_features_tasks = []
 for extract_load_task_id in range(1, extract_load_tasks_count + 1):
     extract_load_features_tasks.append(extract_load_features.override(task_id = f"extraire_charger_features_{extract_load_task_id}")(extract_load_task_id))
@@ -428,7 +428,7 @@ compute_cells_task >> extract_load_features_tasks
 
 Après chargement, la table définitive de stockage des données est écrasée avec la table de chargement. Seul le champ `informations` est conservé, les autres champs n'étant utiles que pour la phase d'EL.
 
-```py
+```py title="Tâche de remplacement de la table destination par la table temporaire de chargement"
 replace_table_task = postgresql_tasks.execute_sql_statement.override(task_id = "remplacer_table")(
     taradata_storage,
     """
@@ -450,7 +450,7 @@ Le passage par la table temporaire `tmp_features` couplé à l'utilisation de la
 
 L'ensemble des tâches est encapsulé dans un _DAG_ planifié de façon mensuelle.
 
-```py
+```py title="Entête du _DAG_ d'extraction/chargement des _features_ Mapillary"
 @dag(dag_id = "extraction_et_chargement__mensuel__mapillary_com",
      start_date = datetime(1993, 1, 10),
      schedule_interval = schedule.get_dag_cron(tags.extract_and_load, tags.monthly),