Get several geojson for UDIs; rename it as atlasante source; update dbt models

Author: jereze

dbt_/models/sources/__sources.yml

@@ -15,8 +15,9 @@ sources:
       - name: laposte_communes
       - name: cog_communes
       - name: opendatasoft_communes
-  - name: udi
+  - name: atlasante
     database: data
     schema: main
-    tables: 
-      - name: atlasante_udi
+    tables:
+      - name: atlasante_udi_2023
+      - name: atlasante_udi_corse

dbt_/models/staging/atlasante/_atlasante_models.yml

@@ -0,0 +1,33 @@
+version: 2
+
+models:
+  - name: stg_atlasante_udi_2023
+    description: "Unités de distribution (UDI) de la France métropolitaine"
+    columns:
+      - name: gid
+        type: INTEGER
+      - name: code_udi
+        type: VARCHAR
+      - name: ins_nom
+        type: VARCHAR
+      - name: uge_nom
+        type: VARCHAR
+      - name: udi_pop
+        type: VARCHAR
+      - name: geom
+        type: GEOMETRY
+      - name: ingestion_date
+        type: DATE
+  - name: stg_atlasante_udi_corse
+    description: "Unités de distribution (UDI) de la Corse"
+    columns:
+      - name: gid
+        type: INTEGER
+      - name: cn_udi
+        type: VARCHAR
+      - name: nom_udi
+        type: VARCHAR
+      - name: geom
+        type: GEOMETRY
+      - name: ingestion_date
+        type: DATE

dbt_/models/staging/atlasante/stg_atlasante_udi_2023.sql

@@ -0,0 +1,9 @@
+SELECT
+    gid::INTEGER AS gid,
+    code_udi::VARCHAR AS code_udi,
+    ins_nom::VARCHAR AS ins_nom,
+    uge_nom::VARCHAR AS uge_nom,
+    udi_pop::VARCHAR AS udi_pop,
+    geom::GEOMETRY AS geom,
+    ingestion_date::DATE AS ingestion_date
+FROM {{ source('atlasante', 'atlasante_udi_2023') }}

dbt_/models/staging/atlasante/stg_atlasante_udi_corse.sql

@@ -0,0 +1,7 @@
+SELECT
+    gid::INTEGER AS gid,
+    cn_udi::VARCHAR AS cn_udi,
+    nom_udi::VARCHAR AS nom_udi,
+    geom::GEOMETRY AS geom,
+    ingestion_date::DATE AS ingestion_date
+FROM {{ source('atlasante', 'atlasante_udi_corse') }}

dbt_/models/staging/udi/_udi_models.yml

@@ -6,13 +6,13 @@
     - custom-years (str) : List of years to process when refresh_type is "custom"
     - drop-tables        : Drop all table before ingestion if flag is added
     - check-update       : For edc, only refresh table if source was updated since last refresh
-    - refresh-table (str): Source to refresh ("all", "edc", "commune", "udi")
+    - refresh-table (str): Source to refresh ("all", "edc", "commune", "atlasante")
 
 Examples:
     - build_database --refresh-table all : refresh all tables
-    - build_database --refresh-table edc : only refresh edc table
-    - build_database --refresh-table commune : only refresh commune table
-    - build_database --refresh-table udi : only refresh udi table
+    - build_database --refresh-table edc : only refresh edc tables
+    - build_database --refresh-table commune : only refresh commune tables
+    - build_database --refresh-table atlasante : only refresh tables from atlasante
     - build_database --refresh-type all : Process all years
     - build_database --refresh-type last : Process last year only
     - build_database --refresh-type custom --custom-years 2018,2024 : Process only the years 2018 and 2024
@@ -28,10 +28,10 @@
 from pipelines.tasks.client.core.duckdb_client import DuckDBClient
 from pipelines.tasks.client.datagouv_client import DataGouvClient
 from pipelines.tasks.client.opendatasoft_client import OpenDataSoftClient
-from pipelines.tasks.client.udi_client import UDIClient
+from pipelines.tasks.client.uploaded_geojson_client import UploadedGeoJSONClient
 from pipelines.tasks.config.config_insee import get_insee_config
 from pipelines.tasks.config.config_laposte import get_laposte_config
-from pipelines.tasks.config.config_udi import get_udi_config
+from pipelines.tasks.config.config_uploaded_geojson import uploaded_geojson_config
 from pipelines.utils.logger import get_logger
 
 logger = get_logger(__name__)
@@ -48,7 +48,7 @@ def execute(
     Execute the EDC dataset processing with specified parameters.
 
     :param refresh_type: Type of refresh to perform ("all", "last", or "custom")
-    :param refresh_table: which table to refresh ("all", "edc","commune", "udi")
+    :param refresh_table: which tables to refresh ("all", "edc", "commune", "atlasante")
     :param custom_years: List of years to process when refresh_type is "custom"
     :param drop_tables: Whether to drop edc tables in the database before data insertion.
     """
@@ -74,8 +74,8 @@ def execute(
         laposte.process_datasets()
         opendatasoft = OpenDataSoftClient(duckdb_client)
         opendatasoft.process_datasets()
-    if refresh_table == "all" or refresh_table == "udi":
-        udi_client = UDIClient(get_udi_config(), duckdb_client)
-        udi_client.process_datasets()
+    if refresh_table == "all" or refresh_table == "atlasante":
+        geojson_client = UploadedGeoJSONClient(uploaded_geojson_config, duckdb_client)
+        geojson_client.process_datasets()
 
     duckdb_client.close()

dbt_/models/staging/udi/stg_udi.sql

@@ -0,0 +1,71 @@
+import os
+from pathlib import Path
+
+from pipelines.tasks.client.core.duckdb_client import DuckDBClient
+from pipelines.tasks.config.common import (
+    CACHE_FOLDER,
+    logger,
+)
+from pipelines.utils.storage_client import ObjectStorageClient
+
+
+class UploadedGeoJSONClient:
+    """Client pour télécharger et ingérer plusieurs fichiers GeoJSON uploadés préalablement manuellement sur S3"""
+
+    def __init__(self, config, duckdb_client: DuckDBClient):
+        self.config = config
+        self.duckdb_client = duckdb_client
+        self.storage_client = ObjectStorageClient()
+
+        if "files" not in self.config:
+            raise ValueError(
+                "Configuration must contain a 'files' list with the GeoJSON files to process"
+            )
+
+        self.files_config = self.config["files"]
+        logger.info(
+            f"UploadedGeoJSONClient initialized with {len(self.files_config)} file(s)"
+        )
+
+    def process_datasets(self):
+        logger.info(f"Processing {self.__class__.__name__} data")
+        self._download_data()
+        self._ingest_to_duckdb()
+        logger.info(f"Finishing processing {self.__class__.__name__} data")
+
+    def _download_data(self):
+        os.makedirs(CACHE_FOLDER, exist_ok=True)
+
+        for file_config in self.files_config:
+            s3_key = (
+                f"{self.config['source'].get('prefix', 'upload')}/{file_config['path']}"
+            )
+            local_path = Path(CACHE_FOLDER, file_config["local_file_name"])
+            logger.info(f"Downloading {s3_key} to {local_path}")
+            self.storage_client.download_object(
+                file_key=s3_key, local_path=str(local_path)
+            )
+
+    def _ingest_to_duckdb(self):
+        logger.info(
+            f"Ingesting {len(self.files_config)} uploaded GeoJSON file(s) into DuckDB"
+        )
+
+        # Collect all table names for dropping
+        table_names = [file_config["table_name"] for file_config in self.files_config]
+        self.duckdb_client.drop_tables(table_names=table_names)
+
+        # Ingest each file
+        for file_config in self.files_config:
+            logger.info(
+                f"Ingesting {file_config['local_file_name']} into table {file_config['table_name']}"
+            )
+            self.duckdb_client.ingest_from_geojson(
+                table_name=file_config["table_name"],
+                filepath=Path(CACHE_FOLDER, file_config["local_file_name"]),
+            )
+            logger.info(
+                f"✅ {file_config['local_file_name']} has been ingested into table {file_config['table_name']}"
+            )
+
+        logger.info("✅ All uploaded GeoJSON files have been ingested in DB")