Use ogr2ogr instead of geopandas

.github/workflows/build-airflow-trixie-image.yml

@@ -0,0 +1,41 @@
+name: build-airflow-base-image.yml
+on:
+  workflow_dispatch:
+  push:
+    paths:
+      - 'docker/airflow-base/**'
+
+jobs:
+  build-airflow:
+    runs-on: ubuntu-latest
+    name: Build & push airflow base
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Getting image tag
+        id: version
+        run: echo "VERSION=$(echo $GITHUB_REF | cut -d / -f 3)" >> $GITHUB_OUTPUT
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Build airflow-base
+        run: |
+          make build-airflow-base
+
+      - name: "Log in to docker.io"
+        if: github.repository == 'georchestra/datafeeder'
+        uses: docker/login-action@v3
+        with:
+          username: '${{ secrets.DOCKER_HUB_USERNAME }}'
+          password: '${{ secrets.DOCKER_HUB_PASSWORD }}'
+
+      - name: Push airflow
+        run: docker push georchestra/airflow-base:latest
+
+      - name: Push release
+        if: github.ref_type == 'tag'
+        run: |
+          docker tag georchestra/airflow-base:latest georchestra/airflow-base:${{ steps.version.outputs.VERSION }}
+          docker push georchestra/airflow-base:${{ steps.version.outputs.VERSION }}

.python-version

@@ -1 +1 @@
-3.12
+3.13

ARCHITECTURE.md

@@ -47,8 +47,8 @@ Supported source types: `FILE`, `URL`, `FTP`, `DATABASE`, `API` (WFS / OGC API F
 ### Shared Library (`libs/data_manipulation/`)
 
 Python package imported by both the backend and the ELT DAGs. Contains:
-- Source-specific ingestion functions (file, URL, FTP, database, OGC services)
-- Transformation pipeline (column remapping, projection reprojection, geometry handling, SQL filters)
+- Source-specific ingestion functions that stream into PostGIS via `ogr2ogr`/GDAL (file, URL, FTP, database, OGC services)
+- SQL-native transformation pipeline (column remapping, projection, geometry handling, filters) executed server-side — data never leaves PostgreSQL except a bounded preview
 - GeoServer write helpers
 - Shared models (`IntegrityTransformation`, column config, etc.)
 

Makefile

@@ -1,6 +1,14 @@
 # Display help message by default
 default: help
 
+# Apache Airflow base image (built locally on Debian Trixie from the official
+# Airflow Dockerfile, since apache/airflow only ships bookworm based images).
+AIRFLOW_VERSION ?= 3.2.2
+AIRFLOW_PYTHON_VERSION ?= 3.13.5
+AIRFLOW_BASE_IMAGE ?= georchestra/airflow-base:$(AIRFLOW_VERSION)-trixie
+export AIRFLOW_VERSION
+export AIRFLOW_BASE_IMAGE
+
 help: ## Display this help message
 	@echo "Usage: make <target>"
 	@echo
@@ -28,6 +36,20 @@ test-backend-coverage: install-python ## Run backend tests with coverage report
 build-libs: install-python ## Build all shared libraries
 	uv build libs/data_manipulation
 
+build-airflow-base: ## Build the Debian Trixie based Apache Airflow base image (from the official Dockerfile)
+	@if [ -z "$$(docker images -q $(AIRFLOW_BASE_IMAGE))" ]; then \
+	  echo "Building $(AIRFLOW_BASE_IMAGE) (Airflow $(AIRFLOW_VERSION), Python $(AIRFLOW_PYTHON_VERSION) on debian:trixie-slim)..."; \
+	  docker build \
+	    --build-arg BASE_IMAGE=debian:trixie-slim \
+	    --build-arg AIRFLOW_VERSION=$(AIRFLOW_VERSION) \
+	    --build-arg AIRFLOW_PYTHON_VERSION=$(AIRFLOW_PYTHON_VERSION) \
+	    -t $(AIRFLOW_BASE_IMAGE) \
+	    docker/airflow-base; \
+	  docker tag $(AIRFLOW_BASE_IMAGE) georchestra/airflow-base:latest; \
+	else \
+	  echo "$(AIRFLOW_BASE_IMAGE) already present, skipping (run 'docker rmi $(AIRFLOW_BASE_IMAGE)' to rebuild)."; \
+	fi
+
 up: build-libs ## Start all services including GeoServer and GeoNetwork using Docker Compose
 	docker compose --profile geoserver --profile geonetwork up -d --wait --build
 
@@ -42,5 +64,5 @@ run-backend: install-python ## Run the backend application
 	DATAFEEDER_CONFIG="$(CURDIR)/apps/backend/datafeeder.env" sh -c \
 	  'uv run alembic upgrade head && uv run uvicorn src.main:app --host 0.0.0.0 --port 8000 --reload --reload-dir ../../apps/backend --reload-dir ../../libs'
 
-.PHONY: default help install-python fix-and-check-all-python build-libs up down down-v run-backend
+.PHONY: default help install-python fix-and-check-all-python build-libs build-airflow-base up down down-v run-backend
 

apps/backend/.python-version

@@ -1 +1 @@
-3.12
+3.13

apps/backend/Dockerfile

@@ -8,7 +8,7 @@ FROM ghcr.io/astral-sh/uv:${UV_VERSION} AS uv
 # ============================================================================
 # Base stage: Common setup for all targets
 # ============================================================================
-FROM python:3.12-slim AS base
+FROM python:3.13-slim AS base
 
 # Install required dependencies with APT
 RUN apt-get update && \
@@ -23,15 +23,15 @@ COPY --from=uv /uv /uvx /bin/
 # - Silence uv complaining about not being able to use hard links,
 # - tell uv to byte-compile packages for faster application startups,
 # - prevent uv from accidentally downloading isolated Python builds,
-# - pick a Python (use `/usr/bin/python3.12` on uv 0.5.0 and later),
+# - pick a Python (use `/usr/bin/python3.13` on uv 0.5.0 and later),
 # - declare `/app` as the target for `uv sync`.
 # - declare a cache directory for uv to use
 # - disable uv caching to ensure fresh installs in CI/CD environments
 # - enable frozen installs to ensure reproducible builds
 ENV UV_LINK_MODE=copy \
     UV_COMPILE_BYTECODE=1 \
     UV_PYTHON_DOWNLOADS=never \
-    UV_PYTHON=python3.12 \
+    UV_PYTHON=python3.13 \
     UV_CACHE_DIR=/tmp/uv-cache \
     UV_NO_CACHE=1 \
     UV_FROZEN=1

apps/backend/pyproject.toml

@@ -15,7 +15,7 @@ name = "datafeeder-backend"
 version = "0.1.0"
 description = "REST API for data ingestion in Datafeeder"
 readme = "README.md"
-requires-python = "==3.12.*"
+requires-python = "==3.13.*"
 dependencies = [
     "fastapi[standard]==0.136.1",
     "uvicorn[standard]==0.46.0",
@@ -28,9 +28,8 @@ dependencies = [
     "psycopg[binary]==3.3.4",
     "geoservercloud",
     "geonetwork",
-    "apache-airflow-client==3.1.8",
+    "apache-airflow-client==3.2.2",
     "geojson-pydantic==2.1.1",
-    "chardet==7.4.3",
     "requests==2.33.1",
     "alembic==1.18.4",
 ]

apps/backend/src/api/routes/ingestion/staging.py

@@ -1,28 +1,24 @@
-import json
 import re
 from datetime import datetime, timezone
 from typing import Any, Optional
 from urllib.parse import urlparse
 from uuid import UUID, uuid4
 
-import geopandas as gpd
-import pandas as pd
 import requests
 from airflow_client.client.models.dag_run_patch_body import DAGRunPatchBody
 from data_manipulation import (
     IntegrityTransformation,
     detect_column_type_from_sqla,
-    read_and_transform_data,
+    detect_table_srid,
+    read_transformed_preview,
 )
 from data_manipulation.constants import DB_URI_PREFIX
 from data_manipulation.database import schema_exists, table_exists
-from data_manipulation.ingestion import read_data_from_postgis
 from data_manipulation.logging import configure_logging
 from data_manipulation.models import ForceProjection as DataManipulationForceProjection
 from data_manipulation.utils import sanitize_name
 from data_manipulation.validators import validate_schema_name, validate_table_name
 from fastapi import APIRouter, Body, File, Form, Header, HTTPException, Query, UploadFile
-from shapely.geometry.base import BaseGeometry
 from sqlalchemy import MetaData, Table, func, select
 from sqlalchemy.orm.attributes import flag_modified
 
@@ -771,9 +767,7 @@ def _detect_original_projection(
 ) -> str | None:
     """Return the CRS string if the staging table contains geographic data."""
     try:
-        sample = read_data_from_postgis(staging_table_name, engine, schema, limit=1)
-        if isinstance(sample, gpd.GeoDataFrame) and sample.crs is not None:
-            return sample.crs.to_string()
+        return detect_table_srid(staging_table_name, engine, schema)
     except Exception as e:
         logger.warning(f"Could not detect original projection: {e}")
     return None
@@ -1060,7 +1054,7 @@ def get_staging_preview(
             config = IntegrityTransformation.model_validate(integrity_link.integrity_transformation)
         except Exception as e:
             logger.warning(f"Could not deserialize transformation config, using raw: {e}")
-    # SECURITY NOTE: when raw=True, config remains None and read_and_transform_data
+    # SECURITY NOTE: when raw=True, config remains None and read_transformed_preview
     # returns ALL columns including those marked as excluded in the saved config.
     # This is intentional — raw mode is a debug/fallback path used when the
     # transformation itself causes a preview error. If excluded columns contain
@@ -1099,79 +1093,22 @@ def get_staging_preview(
         )
 
     try:
-        transformed_data = read_and_transform_data(
-            staging_table_name, engine, schema, config, limit=limit
+        preview = read_transformed_preview(
+            staging_table_name, engine, config, schema=schema, limit=limit
         )
 
-        # Convert all non-JSON-serializable types to string (datetime, Timestamp, etc.)
-        for col in transformed_data.columns:
-            if transformed_data[col].dtype == "object":
-                try:
-                    if pd.api.types.is_datetime64_any_dtype(transformed_data[col]):
-                        transformed_data[col] = transformed_data[col].astype(str)  # type: ignore[misc]
-                except Exception:
-                    pass
-            elif pd.api.types.is_datetime64_any_dtype(transformed_data[col]):
-                transformed_data[col] = transformed_data[col].astype(str)  # type: ignore[misc]
-
-        data: list[dict[str, Any]] = []
-        geojson_data = None
-        is_geographic = False
-
-        # Convert geometry to WKT for tabular display if GeoDataFrame
-        if isinstance(transformed_data, gpd.GeoDataFrame):
-            is_geographic = True
-
-            geometry_cols: list[str] = []
-            for col in transformed_data.columns:  # type: ignore[misc]
-                if not transformed_data[col].empty:  # type: ignore[misc]
-                    sample_item = transformed_data[col].iloc[0]
-                    sample: Any = sample_item  # type: ignore[misc]
-
-                    if isinstance(sample, BaseGeometry):
-                        geometry_cols.append(col)  # type: ignore[misc]
-                    elif hasattr(sample_item, "wkt"):  # type: ignore[misc]
-                        geometry_cols.append(col)  # type: ignore[misc]
-
-            logger.info(f"Found geometry columns: {geometry_cols}")
-
-            # Create GeoJSON for map display first, force to EPSG:4326
-            map_gdf = transformed_data.copy()
-
-            try:
-                if map_gdf.crs and map_gdf.crs.to_string() != "EPSG:4326":
-                    map_gdf = map_gdf.to_crs("EPSG:4326")
-                    logger.info(f"Reprojected data from {map_gdf.crs} to EPSG:4326 for map display")
-            except Exception as crs_error:
-                logger.warning(f"Could not reproject to EPSG:4326: {crs_error}")
-
-            # Modify transformed_data directly for tabular display
-            if "geom" in geometry_cols:
-                transformed_data["geom"] = transformed_data["geom"].apply(  # type: ignore[misc]
-                    lambda geom: geom.wkt if geom is not None else None  # type: ignore[misc]
-                )
-                geometry_cols.remove("geom")
-
-            # Drop extra geometry columns for tabular data
-            table_data = transformed_data.drop(columns=geometry_cols, errors="ignore")
-            data = table_data.to_dict(orient="records")  # type: ignore[misc]
-
-            geojson_str = map_gdf.to_json()  # type: ignore[misc]
-            geojson_data = json.loads(geojson_str) if geojson_str else None
-        else:
-            # Regular DataFrame, no geometry conversion needed
-            data = transformed_data.to_dict(orient="records")  # type: ignore[misc]
+        data = preview.rows
+        geojson_data = preview.geojson
+        is_geographic = preview.is_geographic
 
         # If the geom column was excluded in the saved config, suppress map data
         # regardless of include_excluded. Raw mode bypasses this rule.
         if geom_excluded_in_config:
             is_geographic = False
             geojson_data = None
 
-        return StagingPreviewResponse(
-            data=data,  # type: ignore[misc]
-            geojson=geojson_data,
-            is_geographic=is_geographic,
+        return StagingPreviewResponse.model_validate(
+            {"data": data, "geojson": geojson_data, "is_geographic": is_geographic}
         )
 
     except Exception as e:

apps/elt/.python-version

@@ -1 +1 @@
-3.12
+3.13

apps/elt/dags/task_groups/ingestion.py

@@ -273,6 +273,29 @@ def api_ingest_step(**context: dict[str, Any]) -> None:
             if not source_layer:
                 raise AirflowException("source_layer is required for API import")
 
+            # Decrypt Basic Auth credentials if provided (e.g. protected WFS/OAPIF services)
+            auth = None
+            encrypted_credentials = params.get("encrypted_credentials")
+            if encrypted_credentials:
+                try:
+                    encryption_key = Variable.get("datafeeder_encryption_key", default=None)
+                    if not encryption_key:
+                        raise AirflowException(
+                            "Encryption key not found in Airflow Variables under 'datafeeder_encryption_key'"
+                        )
+
+                    datafeeder_engine = get_datafeeder_sql_engine()
+
+                    with datafeeder_engine.connect() as conn:
+                        username, password = decrypt_credentials(
+                            conn, encrypted_credentials, encryption_key
+                        )
+                        auth = (username, password)
+                        logger.info("Successfully decrypted Basic Auth credentials")
+                except Exception as e:
+                    logger.error(f"Failed to decrypt Basic Auth credentials: {e}")
+                    raise AirflowException(f"Failed to decrypt credentials: {e}")
+
             engine = get_data_sql_engine()
             try:
                 ingest_data_from_ogc_service_into_postgis(
@@ -282,6 +305,7 @@ def api_ingest_step(**context: dict[str, Any]) -> None:
                     target_table_name,
                     engine,
                     schema=get_staging_schema(),
+                    auth=auth,
                 )
             except Exception as e:
                 raise AirflowException(f"Failed to ingest data from OGC service: {e}")

apps/elt/dags/task_groups/transformation.py

@@ -8,8 +8,7 @@
 from airflow.utils.trigger_rule import TriggerRule
 from data_manipulation import (
     IntegrityTransformation,
-    read_and_transform_data,
-    write_data_to_postgis,
+    transform_staging_to_final,
 )
 from data_manipulation.database import create_schema
 from sqlalchemy import MetaData, Table
@@ -82,18 +81,6 @@ def read_transform_write_task(**context: dict[str, Any]) -> None:
                 logger.info(
                     f"Reading and transforming data from {staging_schema}.{staging_table_name}"
                 )
-                transformed_data = read_and_transform_data(
-                    table_name=staging_table_name,
-                    engine=engine,
-                    schema=staging_schema,
-                    config=transformation_config,
-                    limit=None,
-                )
-                logger.info(f"Transformations applied to {len(transformed_data)} rows")
-
-                if transformed_data.empty:
-                    logger.error("No data to write after transformation.")
-                    raise AirflowException("No data to write after transformation.")
 
                 # If this is a re-run (has last_retrieval_timestamp), drop the old final table first
                 last_retrieval_timestamp = params.get("last_retrieval_timestamp")
@@ -118,16 +105,30 @@ def read_transform_write_task(**context: dict[str, Any]) -> None:
                         )
 
                 create_schema(engine, final_schema)
-                logger.info(f"Writing data to {final_schema}.{final_table_name}")
-                write_data_to_postgis(
-                    data=transformed_data,
-                    table_name=final_table_name,
+                logger.info(
+                    f"Transforming {staging_schema}.{staging_table_name} "
+                    f"into {final_schema}.{final_table_name}"
+                )
+                # Transformation runs entirely in PostGIS (CREATE TABLE AS) — no
+                # data is loaded into Python memory.
+                row_count = transform_staging_to_final(
+                    staging_table=staging_table_name,
+                    final_table=final_table_name,
                     engine=engine,
-                    schema=final_schema,
+                    config=transformation_config,
+                    staging_schema=staging_schema,
+                    final_schema=final_schema,
                     create_id=True,
                 )
-                logger.info(f"Successfully wrote {len(transformed_data)} rows to final table")
 
+                if row_count == 0:
+                    logger.error("No data to write after transformation.")
+                    raise AirflowException("No data to write after transformation.")
+
+                logger.info(f"Successfully wrote {row_count} rows to final table")
+
+            except AirflowException:
+                raise
             except Exception as e:
                 raise AirflowException(f"Failed to transform and load data: {e}")