WIP: [draft] add event listener to link stations to operational units

src/api/qualicharge/factories/static.py

@@ -56,8 +56,11 @@ class StatiqueFactory(ModelFactory[Statique]):
     )
     date_maj = Use(DataclassFactory.__faker__.past_date)
     date_mise_en_service = Use(DataclassFactory.__faker__.past_date)
+    # id_station_itinerance = Use(
+    #     FrenchDataclassFactory.__faker__.pystr_format, "FR###P######"
+    # )
     id_station_itinerance = Use(
-        FrenchDataclassFactory.__faker__.pystr_format, "FR###P######"
+        FrenchDataclassFactory.__faker__.pystr_format, "FRZUNP######"
     )
     id_pdc_itinerance = Use(
         FrenchDataclassFactory.__faker__.pystr_format, "FR###E######"
@@ -126,7 +129,8 @@ class StationFactory(TimestampedSQLModelFactory[Station]):
     """Station schema factory."""
 
     id_station_itinerance = Use(
-        FrenchDataclassFactory.__faker__.pystr_format, "FR###P######"
+        # FrenchDataclassFactory.__faker__.pystr_format, "FR###P######"
+        FrenchDataclassFactory.__faker__.pystr_format, "FRZUNP######"
     )
     date_maj = Use(DataclassFactory.__faker__.past_date)
     date_mise_en_service = Use(DataclassFactory.__faker__.past_date)

src/api/qualicharge/schemas/__init__.py

@@ -19,14 +19,16 @@
     field_validator,
 )
 from pydantic_extra_types.coordinate import Coordinate
+from qualicharge.exceptions import ObjectDoesNotExist
 from shapely.geometry import mapping
-from sqlalchemy import CheckConstraint
+from sqlalchemy import CheckConstraint, event
 from sqlalchemy.schema import Column as SAColumn
 from sqlalchemy.types import Date, DateTime, String
 from sqlmodel import Field, Relationship, SQLModel, UniqueConstraint, select
 from sqlmodel import Session as SMSession
 from sqlmodel.main import SQLModelConfig
 
+from qualicharge.db import get_session
 from ..models.dynamic import SessionBase, StatusBase
 from ..models.static import (
     AccessibilitePMREnum,
@@ -289,6 +291,21 @@ def __eq__(self, other) -> bool:
         return all(getattr(self, field) == getattr(other, field) for field in fields)
 
 
+@event.listens_for(Station, "before_insert")
+@event.listens_for(Station, "before_update")
+def link_station_to_operational_unit(mapper, connection, target):
+    """Automatically link station to an operational unit."""
+    code = target.id_station_itinerance[:5]
+    operational_unit = connection.execute(
+        select(OperationalUnit).where(OperationalUnit.code == code)
+    ).one_or_none()
+    if operational_unit is None:
+        raise ObjectDoesNotExist(
+            f"OperationalUnit with code {code} should be created first"
+        )
+    target.operational_unit_id = operational_unit.id
+
+
 class PointDeCharge(BaseTimestampedSQLModel, table=True):
     """PointDeCharge table."""