API périmètres - Limiter aux événements avant une date

fix #1251

Author: rik

django/core/models.py

@@ -1,4 +1,5 @@
 import uuid
+from datetime import date
 from enum import StrEnum, auto
 from itertools import groupby
 from json import load
@@ -134,7 +135,11 @@ class EventImpact(StrEnum):
 
 
 class ProcedureQuerySet(models.QuerySet):
-    def with_events(self) -> Self:
+    def with_events(self, *, avant: date | None = None) -> Self:
+        events = Event.objects.all()
+        if avant:
+            events = events.filter(date_evenement_string__lt=str(avant))
+
         approbation_event_types = [
             event_type
             for event_impact_by_event_type in EVENT_IMPACT_BY_TYPE_DOCUMENT.values()
@@ -146,7 +151,7 @@ def with_events(self) -> Self:
             models.When(
                 type_document=type_document,
                 then=models.Subquery(
-                    Event.objects.filter(
+                    events.filter(
                         procedure=models.OuterRef("pk"),
                         is_valid=True,
                         type__in=event_impact_by_event_type.keys(),
@@ -162,7 +167,7 @@ def with_events(self) -> Self:
         return self.annotate(
             date_approbation=Coalesce(
                 models.Subquery(
-                    Event.objects.filter(
+                    events.filter(
                         procedure=models.OuterRef("pk"),
                         type__in=approbation_event_types,
                     ).values("date_evenement_string")[:1]
@@ -271,13 +276,16 @@ def with_opposabilite(
         departement: str | None = None,
         collectivite_code: str | None = None,
         collectivite_type: str | None = None,
+        avant: date | None = None,
     ) -> list["CommuneProcedure"]:
         communes_procedures = (
             self.filter(
                 procedure__is_principale=True,
                 procedure__archived=False,
             )
-            .prefetch_related(models.Prefetch("procedure", Procedure.objects.all()))
+            .prefetch_related(
+                models.Prefetch("procedure", Procedure.objects.with_events(avant=avant))
+            )
             .order_by("collectivite_code", "collectivite_type")
         )
 

django/core/tests/test_models.py

@@ -1,3 +1,5 @@
+from datetime import date
+
 import pytest
 from pytest_django import DjangoAssertNumQueries
 
@@ -94,6 +96,31 @@ def test_date_approbation_quand_event_approbation_manquant(
 
             assert procedure_with_events.date_approbation == "0000-00-00"
 
+    @pytest.mark.django_db
+    def test_date_approbation_ignore_event_apres(
+        self, django_assert_num_queries: DjangoAssertNumQueries
+    ) -> None:
+        procedure = Procedure.objects.create(type_document=TypeDocument.PLUI)
+        procedure.event_set.create(
+            type="Caractère exécutoire",
+            date_evenement_string="2022-12-01",
+        )
+        procedure.event_set.create(
+            type="Délibération d'approbation",
+            date_evenement_string="2023-12-01",
+        )
+
+        assert [event.impact for event in procedure.event_set.all()] == [
+            EventImpact.OPPOSABLE,
+            EventImpact.OPPOSABLE,
+        ]
+        with django_assert_num_queries(1):
+            procedure_with_events = Procedure.objects.with_events(
+                avant=date(2023, 12, 1)
+            ).get(id=procedure.id)
+
+            assert procedure_with_events.date_approbation == "2022-12-01"
+
 
 class TestProcedureStatut:
     @pytest.mark.django_db
@@ -111,6 +138,42 @@ def test_principale_opposable(
 
             assert procedure_with_events.statut == EventImpact.OPPOSABLE
 
+    @pytest.mark.django_db
+    @pytest.mark.parametrize(
+        ("jour_limite", "impact"),
+        [
+            (3, None),
+            (4, EventImpact.EN_COURS),
+            (5, EventImpact.EN_COURS),
+            (6, EventImpact.OPPOSABLE),
+        ],
+    )
+    def test_ignore_event_apres(
+        self,
+        django_assert_num_queries: DjangoAssertNumQueries,
+        jour_limite: int,
+        impact: EventImpact,
+    ) -> None:
+        procedure = Procedure.objects.create(
+            is_principale=True, type_document=TypeDocument.PLUI
+        )
+        event_prescription = procedure.event_set.create(
+            type="Délibération de prescription du conseil municipal ou communautaire",
+            date_evenement_string="2024-12-03",
+        )
+        event_approbation = procedure.event_set.create(
+            type="Caractère exécutoire", date_evenement_string="2024-12-05"
+        )
+
+        assert event_prescription.impact == EventImpact.EN_COURS
+        assert event_approbation.impact == EventImpact.OPPOSABLE
+        with django_assert_num_queries(1):
+            procedure_with_events = Procedure.objects.with_events(
+                avant=date(2024, 12, jour_limite)
+            ).get(id=procedure.id)
+
+            assert procedure_with_events.statut == impact
+
     @pytest.mark.django_db
     def test_principale_sans_evenement(
         self, django_assert_num_queries: DjangoAssertNumQueries
@@ -454,3 +517,46 @@ def test_ignore_procedures_archivees(
             perimetres = CommuneProcedure.objects.with_opposabilite()
         with django_assert_num_queries(0):
             assert perimetres == [commune_procedure_reelle]
+
+    @pytest.mark.django_db
+    def test_ignore_event_apres(
+        self, django_assert_num_queries: DjangoAssertNumQueries
+    ) -> None:
+        procedure_opposable_fevrier = Procedure.objects.create(
+            is_principale=True, type_document=TypeDocument.PLUI
+        )
+        procedure_opposable_fevrier.event_set.create(
+            type="Caractère exécutoire", date_evenement_string="2024-02-01"
+        )
+        commune_procedure_opposable_fevrier = (
+            procedure_opposable_fevrier.perimetre.create(collectivite_code="12345")
+        )
+
+        procedure_opposable_janvier = Procedure.objects.create(
+            is_principale=True, type_document=TypeDocument.PLU
+        )
+        procedure_opposable_janvier.event_set.create(
+            type="Caractère exécutoire", date_evenement_string="2024-01-01"
+        )
+        commune_procedure_opposable_janvier = (
+            procedure_opposable_janvier.perimetre.create(collectivite_code="12345")
+        )
+
+        with django_assert_num_queries(1):
+            procedures = Procedure.objects.with_events()
+
+            assert all(
+                procedure.statut == EventImpact.OPPOSABLE for procedure in procedures
+            )
+
+        with django_assert_num_queries(2):
+            perimetres = CommuneProcedure.objects.with_opposabilite(
+                avant=date(2024, 2, 1)
+            )
+        with django_assert_num_queries(0):
+            assert perimetres == [
+                commune_procedure_opposable_fevrier,
+                commune_procedure_opposable_janvier,
+            ]
+            assert not perimetres[0].opposable
+            assert perimetres[1].opposable

django/core/tests/test_views.py

@@ -57,3 +57,28 @@ def test_retourne_tout_sans_filtre_departement(self, client: Client) -> None:
         response = client.get("/api/perimetres")
         reader = DictReader(response.content.decode().splitlines())
         assert len(list(reader)) == 2
+
+    @pytest.mark.django_db
+    def test_ignore_event_apres(self, client: Client) -> None:
+        commune_procedure = create_commune_procedure(code="12345", departement="12")
+        commune_procedure.procedure.event_set.create(
+            type="Caractère exécutoire", date_evenement_string="2023-01-01"
+        )
+
+        response = client.get("/api/perimetres", {"avant": "2023-01-01"})
+        reader = DictReader(response.content.decode().splitlines())
+        assert [cp["opposable"] for cp in reader] == ["False"]
+
+    @pytest.mark.django_db
+    @pytest.mark.parametrize(
+        "invalid_avant",
+        ["2023-1-01", "2023-02-30", "invalid-date", "2023/01/01"],
+    )
+    def test_parsing_avant(self, client: Client, invalid_avant: str) -> None:
+        response = client.get("/api/perimetres", {"avant": invalid_avant})
+
+        assert response.status_code == 400
+        assert (
+            response.content.decode()
+            == "Le paramètre 'avant' doit être une date valide au format YYYY-MM-DD."
+        )

django/core/views.py

@@ -1,7 +1,8 @@
 from csv import DictWriter
+from datetime import date
 from operator import attrgetter
 
-from django.http import HttpRequest, HttpResponse
+from django.http import HttpRequest, HttpResponse, HttpResponseBadRequest
 from django.shortcuts import render
 from django.views.decorators.http import require_safe
 
@@ -11,8 +12,20 @@
 @require_safe
 def api_perimetres(request: HttpRequest) -> HttpResponse:
     departement = request.GET.get("departement")
+
+    try:
+        avant = (
+            date.fromisoformat(request.GET.get("avant"))
+            if request.GET.get("avant")
+            else None
+        )
+    except ValueError:
+        return HttpResponseBadRequest(
+            "Le paramètre 'avant' doit être une date valide au format YYYY-MM-DD."
+        )
+
     communes_procedures = CommuneProcedure.objects.with_opposabilite(
-        departement=departement
+        departement=departement, avant=avant
     )
 
     response = HttpResponse(content_type="text/csv;charset=utf-8")

nuxt/content/Dev/API/urba/perimetres.md

@@ -20,6 +20,7 @@ Retourne les procédures de chaque commune et leur opposabilité.
 #### Paramètres de requête disponibles
 
 - `departement` : Filtrer par le code INSEE du département.
+- `avant` : Ne prend en compte que les événements avant ce jour, au format `YYYY-MM-DD`.
 
 ### Réponse