WIP: add status list endpoint

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

@@ -6,6 +6,7 @@
 from annotated_types import Len
 from fastapi import APIRouter, Depends, HTTPException, Path
 from fastapi import status as fa_status
+from sqlalchemy import func
 from sqlalchemy.schema import Column as SAColumn
 from sqlmodel import Session, select
 
@@ -35,9 +36,38 @@
 
 
 @router.get("/status/", tags=["Status"])
-async def list_statuses() -> List[StatusRead]:
+async def list_statuses(
+    session: Session = Depends(get_session),
+) -> List[StatusRead]:
     """List last known point of charge statuses."""
-    raise NotImplementedError
+    # Get latest status per point of charge
+    latest_db_statuses_stmt = (
+        select(
+            Status.point_de_charge_id,
+            func.last(Status.id, Status.horodatage).label("status_id"),
+        )
+        .group_by(cast(SAColumn, Status.point_de_charge_id))
+        .subquery()
+    )
+    db_statuses = session.exec(
+        select(Status).join_from(
+            Status,
+            latest_db_statuses_stmt,
+            Status.id == latest_db_statuses_stmt.c.status_id,  # type: ignore[arg-type]
+        )
+    ).all()
+    return [
+        StatusRead(
+            **s.model_dump(
+                exclude={
+                    "id",
+                    "point_de_charge_id",
+                }
+            )
+        )
+        for s in db_statuses
+        if s is not None
+    ]
 
 
 @router.get("/status/{id_pdc_itinerance}", tags=["Status"])

src/api/qualicharge/factories/__init__.py

@@ -1,9 +1,13 @@
 """QualiCharge factories."""
 
+from datetime import datetime, timedelta, timezone
 from typing import Generic, TypeVar
+from uuid import uuid4
 
 from faker import Faker
+from polyfactory import Use
 from polyfactory.factories.dataclass_factory import DataclassFactory
+from polyfactory.factories.sqlalchemy_factory import SQLAlchemyFactory
 
 T = TypeVar("T")
 
@@ -13,3 +17,20 @@ class FrenchDataclassFactory(Generic[T], DataclassFactory[T]):
 
     __faker__ = Faker(locale="fr_FR")
     __is_base_factory__ = True
+
+
+class TimestampedSQLModelFactory(Generic[T], SQLAlchemyFactory[T]):
+    """A base factory for timestamped SQLModel.
+
+    We expect SQLModel to define the following fields:
+
+    - id: UUID
+    - created_at: datetime
+    - updated_at: datetime
+    """
+
+    __is_base_factory__ = True
+
+    id = Use(uuid4)
+    created_at = Use(lambda: datetime.now(timezone.utc) - timedelta(hours=1))
+    updated_at = Use(datetime.now, timezone.utc)

src/api/qualicharge/factories/dynamic.py

@@ -4,7 +4,8 @@
 from polyfactory.factories.pydantic_factory import ModelFactory
 
 from ..models.dynamic import SessionCreate, StatusCreate
-from . import FrenchDataclassFactory
+from ..schemas import Status
+from . import FrenchDataclassFactory, TimestampedSQLModelFactory
 
 
 class SessionCreateFactory(ModelFactory[SessionCreate]):
@@ -21,3 +22,7 @@ class StatusCreateFactory(ModelFactory[StatusCreate]):
     id_pdc_itinerance = Use(
         FrenchDataclassFactory.__faker__.pystr_format, "FR###E######"
     )
+
+
+class StatusFactory(TimestampedSQLModelFactory[Status]):
+    """Status schema factory."""

src/api/qualicharge/factories/static.py

@@ -1,14 +1,11 @@
 """QualiCharge static factories."""
 
-from datetime import datetime, timedelta, timezone
-from typing import Any, Callable, Dict, Generic, TypeVar
-from uuid import uuid4
+from typing import Any, Callable, Dict, TypeVar
 
 from geoalchemy2.types import Geometry
 from polyfactory import Use
 from polyfactory.factories.dataclass_factory import DataclassFactory
 from polyfactory.factories.pydantic_factory import ModelFactory
-from polyfactory.factories.sqlalchemy_factory import SQLAlchemyFactory
 from pydantic_extra_types.coordinate import Coordinate
 
 from ..models.static import Statique
@@ -20,7 +17,7 @@
     PointDeCharge,
     Station,
 )
-from . import FrenchDataclassFactory
+from . import FrenchDataclassFactory, TimestampedSQLModelFactory
 
 T = TypeVar("T")
 
@@ -66,23 +63,6 @@ class StatiqueFactory(ModelFactory[Statique]):
     )
 
 
-class TimestampedSQLModelFactory(Generic[T], SQLAlchemyFactory[T]):
-    """A base factory for timestamped SQLModel.
-
-    We expect SQLModel to define the following fields:
-
-    - id: UUID
-    - created_at: datetime
-    - updated_at: datetime
-    """
-
-    __is_base_factory__ = True
-
-    id = Use(uuid4)
-    created_at = Use(lambda: datetime.now(timezone.utc) - timedelta(hours=1))
-    updated_at = Use(datetime.now, timezone.utc)
-
-
 class AmenageurFactory(TimestampedSQLModelFactory[Amenageur]):
     """Amenageur schema factory."""
 

src/api/qualicharge/schemas/__init__.py

@@ -13,6 +13,7 @@
     PositiveFloat,
     PositiveInt,
     ValidationInfo,
+    computed_field,
     field_serializer,
     field_validator,
 )
@@ -313,3 +314,9 @@ class Status(BaseTimestampedSQLModel, StatusBase, table=True):
         default=None, foreign_key="pointdecharge.id"
     )
     point_de_charge: PointDeCharge = Relationship(back_populates="statuses")
+
+    @computed_field  # type: ignore[misc]
+    @property
+    def id_pdc_itinerance(self) -> str:
+        """Return the PointDeCharge.id_pdc_itinerance (used for serialization only)."""
+        return self.point_de_charge.id_pdc_itinerance

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

@@ -1,19 +1,79 @@
 """Tests for the QualiCharge API dynamic router."""
 
 import json
-
-from sqlalchemy import func
+from typing import cast
 
 from fastapi import status
+from sqlalchemy import func
+from sqlalchemy.schema import Column as SAColumn
 from sqlmodel import select
 
 from qualicharge.conf import settings
-from qualicharge.factories.dynamic import SessionCreateFactory, StatusCreateFactory
+from qualicharge.factories.dynamic import (
+    SessionCreateFactory,
+    StatusCreateFactory,
+    StatusFactory,
+)
 from qualicharge.factories.static import StatiqueFactory
+from qualicharge.models.dynamic import StatusRead
 from qualicharge.schemas import PointDeCharge, Session, Status
 from qualicharge.schemas.utils import save_statique, save_statiques
 
 
+def test_list_statuses(db_session, client_auth):
+    """Test the /status/ get endpoint."""
+    StatusFactory.__session__ = db_session
+
+    # No status exists
+    response = client_auth.get("/dynamique/status/")
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json() == []
+
+    # Create points of charge and statuses
+    n_pdc = 2
+    n_status_by_pdc = 10
+    list(save_statiques(db_session, StatiqueFactory.batch(n_pdc)))
+    pdcs = db_session.exec(select(PointDeCharge)).all()
+    assert len(pdcs) == n_pdc
+
+    StatusFactory.create_batch_sync(n_status_by_pdc, point_de_charge_id=pdcs[0].id)
+    StatusFactory.create_batch_sync(n_status_by_pdc, point_de_charge_id=pdcs[1].id)
+    assert db_session.exec(select(func.count(Status.id))).one() == (
+        n_pdc * n_status_by_pdc
+    )
+
+    # List latest statuses by pdc
+    response = client_auth.get("/dynamique/status/")
+    assert response.status_code == status.HTTP_200_OK
+    statuses = [StatusRead(**s) for s in response.json()]
+    assert len(statuses) == n_pdc
+
+    # Check status
+    for response_status in statuses:
+        pdc = db_session.exec(
+            select(PointDeCharge).where(
+                PointDeCharge.id_pdc_itinerance == response_status.id_pdc_itinerance
+            )
+        ).one()
+        db_status = db_session.exec(
+            select(Status)
+            .where(Status.point_de_charge_id == pdc.id)
+            .order_by(cast(SAColumn, Status.horodatage).desc())
+        ).first()
+        assert db_status.etat_pdc == response_status.etat_pdc
+        assert db_status.occupation_pdc == response_status.occupation_pdc
+        assert db_status.horodatage == response_status.horodatage.astimezone()
+        assert db_status.etat_prise_type_2 == response_status.etat_prise_type_2
+        assert (
+            db_status.etat_prise_type_combo_ccs
+            == response_status.etat_prise_type_combo_ccs
+        )
+        assert (
+            db_status.etat_prise_type_chademo == response_status.etat_prise_type_chademo
+        )
+        assert db_status.etat_prise_type_ef == response_status.etat_prise_type_ef
+
+
 def test_create_status_for_non_existing_point_of_charge(client_auth):
     """Test the /status/ create endpoint for non existing point of charge."""
     id_pdc_itinerance = "ESZUNE1111ER1"