Switch static read to MV

src/api/qualicharge/api/v1/routers/static.py

@@ -36,11 +36,10 @@
     PermissionDenied,
 )
 from qualicharge.models.static import Statique
-from qualicharge.schemas.core import OperationalUnit, PointDeCharge, Station
+from qualicharge.schemas.core import OperationalUnit, PointDeCharge, Station, StatiqueMV
 from qualicharge.schemas.sql import StatiqueImporter
 from qualicharge.schemas.utils import (
     are_pdcs_allowed_for_user,
-    build_statique,
     is_pdc_allowed_for_user,
     list_statique,
     save_statique,
@@ -181,13 +180,15 @@ async def read(
         raise PermissionDenied("You don't manage this point of charge")
 
     try:
-        statique = build_statique(session, id_pdc_itinerance)
+        statique_mv = session.exec(
+            select(StatiqueMV).where(StatiqueMV.id_pdc_itinerance == id_pdc_itinerance)
+        ).one()
     except ObjectDoesNotExist as err:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
             detail="Requested statique does not exist",
         ) from err
-    return statique
+    return statique_mv.model_dump(exclude={"pdc_id", "pdc_updated_at"})
 
 
 @router.put("/{id_pdc_itinerance}", status_code=status.HTTP_200_OK)

src/api/qualicharge/schemas/core.py

@@ -2,18 +2,23 @@
 
 from datetime import datetime
 from enum import IntEnum
-from typing import TYPE_CHECKING, ClassVar, List, Optional, Union, cast
+from typing import TYPE_CHECKING, Annotated, ClassVar, List, Optional, Union, cast
 from uuid import UUID, uuid4
 
+from geoalchemy2.functions import ST_AsGeoJSON, ST_GeomFromEWKB
 from geoalchemy2.shape import to_shape
 from geoalchemy2.types import Geometry, WKBElement
 from pydantic import (
+    BeforeValidator,
     EmailStr,
     PastDate,
     PastDatetime,
+    PlainSerializer,
+    PlainValidator,
     PositiveFloat,
     PositiveInt,
     ValidationInfo,
+    WrapValidator,
     computed_field,
     field_serializer,
     field_validator,
@@ -494,6 +499,12 @@ class StatiqueMV(Statique, SQLModel):
     pdc_id: UUID
     pdc_updated_at: datetime
 
+    # WKBElement to Coordinate
+    @field_serializer("coordonneesXY")
+    def _wkb_to_coordinates(value: WKBElement):
+        """Convert WKB to Coordinate."""
+        return Localisation._wkb_to_coordinates(value)
+
 
 class _StatiqueMV(SQLModel):
     """Statique Materialized view.

src/api/tests/api/v1/routers/test_statique.py

@@ -10,6 +10,7 @@
 from pydantic_extra_types.coordinate import Coordinate
 from sqlalchemy import Column as SAColumn
 from sqlalchemy import func
+from sqlalchemy_utils import refresh_materialized_view
 from sqlmodel import select
 
 from qualicharge.auth.factories import GroupFactory
@@ -18,6 +19,7 @@
 from qualicharge.factories.static import StatiqueFactory
 from qualicharge.models.static import Statique
 from qualicharge.schemas.core import (
+    STATIQUE_MV_TABLE_NAME,
     OperationalUnit,
     PointDeCharge,
     Station,
@@ -288,6 +290,7 @@ def test_read_for_superuser(client_auth, db_session):
     db_statique = save_statique(
         db_session, StatiqueFactory.build(id_pdc_itinerance=id_pdc_itinerance)
     )
+    refresh_materialized_view(db_session, STATIQUE_MV_TABLE_NAME)
 
     response = client_auth.get(f"/statique/{id_pdc_itinerance}")
     assert response.status_code == status.HTTP_200_OK
@@ -318,6 +321,7 @@ def test_read_for_user(client_auth, db_session):
     db_statique = save_statique(
         db_session, StatiqueFactory.build(id_pdc_itinerance=id_pdc_itinerance)
     )
+    refresh_materialized_view(db_session, STATIQUE_MV_TABLE_NAME)
 
     # User has no assigned operational units
     response = client_auth.get(f"/statique/{id_pdc_itinerance}")

src/api/tests/schemas/test_static.py

@@ -24,6 +24,7 @@
     StatiqueFactory,
 )
 from qualicharge.schemas.core import (
+    STATIQUE_MV_TABLE_NAME,
     Amenageur,
     Localisation,
     OperationalUnit,
@@ -410,7 +411,7 @@ def test_statique_materialized_view(db_session):
     n_pdc = 4
     statiques = StatiqueFactory.batch(n_pdc)
     save_statiques(db_session, statiques)
-    refresh_materialized_view(db_session, "statique")
+    refresh_materialized_view(db_session, STATIQUE_MV_TABLE_NAME)
 
     db_statiques = db_session.exec(select(StatiqueMV)).all()
     assert len(db_statiques) == n_pdc
@@ -421,3 +422,17 @@ def test_statique_materialized_view(db_session):
     assert isinstance(db_statiques[0].coordonneesXY, WKBElement)
     assert isinstance(db_statiques[0].pdc_id, UUID)
     assert isinstance(db_statiques[0].pdc_updated_at, datetime)
+
+
+def test_statique_materialized_view_coordonneesXY_field(db_session):
+    """Test the coordonneesXY field from the Statique schema."""
+    n_pdc = 1
+    statiques = StatiqueFactory.batch(n_pdc)
+    save_statiques(db_session, statiques)
+    refresh_materialized_view(db_session, STATIQUE_MV_TABLE_NAME)
+
+    db_statique = db_session.exec(select(StatiqueMV)).one()
+    assert isinstance(db_statique.coordonneesXY, WKBElement)
+    crds = Coordinate(**db_statique.model_dump()["coordonneesXY"])
+    assert hasattr(crds, "latitude")
+    assert hasattr(crds, "longitude")