WIP

src/api/Pipfile

@@ -29,6 +29,7 @@ sqlalchemy-timescaledb = "==0.4.1"
 sqlmodel = "==0.0.22"
 typer = "==0.15.1"
 uvicorn = {extras = ["standard"] }
+sqlalchemy-utils = "==0.41.2"
 
 [dev-packages]
 black = "==24.10.0"

src/api/qualicharge/schemas/core.py

@@ -4,6 +4,7 @@
 from typing import TYPE_CHECKING, List, Optional, Union, cast
 from uuid import UUID, uuid4
 
+from geoalchemy2.functions import ST_AsGeoJSON
 from geoalchemy2.shape import to_shape
 from geoalchemy2.types import Geometry, WKBElement
 from pydantic import (
@@ -20,9 +21,12 @@
 from pydantic_extra_types.coordinate import Coordinate
 from shapely.geometry import mapping
 from sqlalchemy import event
+from sqlalchemy.orm import registry
 from sqlalchemy.schema import Column as SAColumn
+from sqlalchemy.schema import Index
 from sqlalchemy.types import Date, DateTime, String
-from sqlmodel import Field, Relationship, UniqueConstraint, select
+from sqlalchemy_utils import create_materialized_view
+from sqlmodel import Field, Relationship, SQLModel, UniqueConstraint, select
 from sqlmodel import Session as SMSession
 from sqlmodel.main import SQLModelConfig
 
@@ -37,12 +41,15 @@
     ImplantationStationEnum,
     NotFutureDate,
     RaccordementEnum,
+    Statique,
 )
 from . import BaseTimestampedSQLModel
 
 if TYPE_CHECKING:
     from qualicharge.auth.schemas import Group
 
+mapper_registry = registry()
+
 
 class OperationalUnitTypeEnum(IntEnum):
     """Operational unit types."""
@@ -389,3 +396,88 @@ class Status(BaseTimestampedSQLModel, StatusBase, table=True):
     def id_pdc_itinerance(self) -> str:
         """Return the PointDeCharge.id_pdc_itinerance (used for serialization only)."""
         return self.point_de_charge.id_pdc_itinerance
+
+
+class StatiqueMV(Statique, SQLModel):
+    """Statique Materialized View."""
+
+    __tablename__ = "statique"
+
+
+class _StatiqueMV(SQLModel):
+    """Statique Materialized view.
+
+    NOTE: This is an internal model used **ONLY** for creating the materialized view.
+    """
+
+    __table__ = create_materialized_view(
+        name="statique",
+        selectable=select(
+            # FIXME
+            # Should be the Enum value
+            PointDeCharge.accessibilite_pmr,
+            PointDeCharge.cable_t2_attache,
+            PointDeCharge.gratuit,
+            PointDeCharge.id_pdc_itinerance,
+            PointDeCharge.id_pdc_local,
+            PointDeCharge.observations,
+            PointDeCharge.paiement_acte,
+            PointDeCharge.paiement_autre,
+            PointDeCharge.paiement_cb,
+            PointDeCharge.prise_type_2,
+            PointDeCharge.prise_type_autre,
+            PointDeCharge.prise_type_chademo,
+            PointDeCharge.prise_type_combo_ccs,
+            PointDeCharge.prise_type_ef,
+            PointDeCharge.puissance_nominale,
+            PointDeCharge.reservation,
+            PointDeCharge.restriction_gabarit,
+            PointDeCharge.tarification,
+            Amenageur.contact_amenageur,
+            Amenageur.nom_amenageur,
+            Amenageur.siren_amenageur,
+            Operateur.contact_operateur,
+            Operateur.nom_operateur,
+            Operateur.telephone_operateur,
+            Enseigne.nom_enseigne,
+            Localisation.adresse_station,
+            Localisation.code_insee_commune,
+            # FIXME
+            # Should add `::json -> 'coordinates'` extraction
+            ST_AsGeoJSON(Localisation.coordonneesXY).label("coordonneesXY"),
+            # FIXME
+            # Should be the Enum value
+            Station.condition_acces,
+            Station.date_maj,
+            Station.date_mise_en_service,
+            Station.horaires,
+            Station.id_station_itinerance,
+            Station.id_station_local,
+            # FIXME
+            # Should be the Enum value
+            Station.implantation_station,
+            Station.nbre_pdc,
+            Station.nom_station,
+            Station.num_pdl,
+            # FIXME
+            # Should be the Enum value
+            Station.raccordement,
+            Station.station_deux_roues,
+        )
+        .join(Station, PointDeCharge.station_id == Station.id)
+        .join(Amenageur, Station.amenageur_id == Amenageur.id)
+        .join(Operateur, Station.operateur_id == Operateur.id)
+        .join(Enseigne, Station.enseigne_id == Enseigne.id)
+        .join(Localisation, Station.localisation_id == Localisation.id),
+        metadata=SQLModel.metadata,
+        indexes=[
+            Index("idx_statique_id_pdc_itinerance", "id_pdc_itinerance", unique=True),
+            Index(
+                "idx_statique_code_insee_commune",
+                "code_insee_commune",
+            ),
+        ],
+    )
+
+
+mapper_registry.map_imperatively(StatiqueMV, _StatiqueMV.__table__)

src/api/tests/schemas/test_static.py

@@ -8,6 +8,7 @@
 from pydantic_extra_types.coordinate import Coordinate
 from shapely.geometry import mapping
 from sqlalchemy.exc import IntegrityError
+from sqlalchemy_utils import refresh_materialized_view
 from sqlmodel import select
 
 from qualicharge.factories.static import (
@@ -18,8 +19,16 @@
     OperationalUnitFactory,
     PointDeChargeFactory,
     StationFactory,
+    StatiqueFactory,
 )
-from qualicharge.schemas.core import Amenageur, Localisation, OperationalUnit, Station
+from qualicharge.schemas.core import (
+    Amenageur,
+    Localisation,
+    OperationalUnit,
+    Station,
+    StatiqueMV,
+)
+from qualicharge.schemas.utils import save_statiques
 
 
 @pytest.mark.parametrize(
@@ -392,3 +401,18 @@ def test_operational_unit_create_stations_fk(db_session):
         select(Station).where(Station.operational_unit_id == operational_unit.id)
     ).all()
     assert len(stations) == n_stations + extra_stations
+
+
+def test_statique_materialized_view(db_session):
+    """Test the StatiqueMV schema."""
+    n_pdc = 4
+    statiques = StatiqueFactory.batch(n_pdc)
+    save_statiques(db_session, statiques)
+    refresh_materialized_view(db_session, "statique", concurrently=False)
+
+    db_statiques = db_session.exec(select(StatiqueMV)).all()
+    assert len(db_statiques) == n_pdc
+
+    assert {s.id_pdc_itinerance for s in statiques} == {
+        s.id_pdc_itinerance for s in db_statiques
+    }