feat: étend le référentiel au parc entier et à la suppression des lots

Trois manques que la saisie a fait apparaître :

Le tableau porte une colonne immeuble, la liste ne peut donc plus être
suspendue à un immeuble choisi d'avance : elle renvoie tout le parc, chaque
ligne emportant de quoi nommer son immeuble sans requête de plus.

Un lot qu'aucune ligne de compte rendu ne mentionne ne décrit rien : il
encombre la saisie et doit pouvoir disparaître. La garde est côté serveur —
un lot porteur de revenus ou de dépenses est refusé, ses montants
partiraient avec lui.

Le nom d'usage s'enregistre, et la réponse recalcule les compteurs de
l'immeuble plutôt que de les laisser à zéro : elle remplace l'immeuble dans
les listes du client, qui le croirait vide de lots.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-28 15:41:24 +02:00
parent 9f5f46a93a
commit fdffe27f06
2 changed files with 290 additions and 35 deletions

View File

@@ -5,7 +5,7 @@ DPE, chauffage) et donnent au référentiel une source de vérité indépendante
l'extraction. l'extraction.
""" """
from fastapi import APIRouter, Depends, HTTPException from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import func, select from sqlalchemy import func, select
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
@@ -16,6 +16,8 @@ from ...utils.logements import delta_surface, dpe_echeance, type_en_ecart
from ..schemas.models import ( from ..schemas.models import (
CaracteristiquesBody, CaracteristiquesBody,
CaracteristiquesResponse, CaracteristiquesResponse,
ImmeubleBody,
ImmeubleResponse,
LotReferentielResponse, LotReferentielResponse,
) )
@@ -44,8 +46,9 @@ def _caracteristiques_response(
def _lot_response( def _lot_response(
lot: Lot, lot: Lot,
immeuble_code: str | None, immeuble_code: str | None,
nb_revenus: int = 0, nb_revenus: int,
nb_depenses: int = 0, nb_depenses: int,
immeuble_denomination: str | None = None,
) -> LotReferentielResponse: ) -> LotReferentielResponse:
"""Assemble la ligne de référentiel d'un lot.""" """Assemble la ligne de référentiel d'un lot."""
fiche = lot.caracteristiques fiche = lot.caracteristiques
@@ -55,6 +58,7 @@ def _lot_response(
numero=lot.numero, numero=lot.numero,
immeuble_id=lot.immeuble_id, immeuble_id=lot.immeuble_id,
immeuble_code=immeuble_code, immeuble_code=immeuble_code,
immeuble_denomination=immeuble_denomination,
type_extrait=lot.type, type_extrait=lot.type,
type_effectif=type_effectif(lot), type_effectif=type_effectif(lot),
type_ecart=type_en_ecart(lot.type, fiche.type if fiche else None), type_ecart=type_en_ecart(lot.type, fiche.type if fiche else None),
@@ -64,27 +68,16 @@ def _lot_response(
) )
@router.get( def _requete_lignes():
"/immeubles/{immeuble_id}/lots/referentiel", """Requête des lots avec le nombre de revenus et de dépenses rattachés.
response_model=list[LotReferentielResponse],
)
async def list_lots_referentiel(
immeuble_id: int,
session: Session = Depends(get_session),
) -> list[LotReferentielResponse]:
"""Liste les lots d'un immeuble avec leur fiche de caractéristiques.
- **immeuble_id**: ID de l'immeuble Partagée par la liste et l'enregistrement : une ligne renvoyée après
écriture doit être comptée comme celle du tableau, sinon un lot bien occupé
se met à passer pour inutilisé dès qu'on le décrit.
Les lots sans fiche sont renvoyés avec `caracteristiques` à `null` : le Sous-requêtes corrélées plutôt que des jointures : compter revenus et
tableau de saisie doit montrer les lignes vides autant que les remplies. dépenses dans la même jointure multiplierait les lignes entre elles.
""" """
immeuble = session.get(Immeuble, immeuble_id)
if immeuble is None:
raise HTTPException(status_code=404, detail="Immeuble introuvable.")
# Sous-requêtes corrélées plutôt que des jointures : compter revenus et
# dépenses dans la même jointure multiplierait les lignes entre elles.
nb_revenus = ( nb_revenus = (
select(func.count(Revenu.id)) select(func.count(Revenu.id))
.where(Revenu.lot_id == Lot.id) .where(Revenu.lot_id == Lot.id)
@@ -98,16 +91,41 @@ async def list_lots_referentiel(
.scalar_subquery() .scalar_subquery()
) )
return select(Lot, nb_revenus.label("nb_revenus"), nb_depenses.label("nb_depenses"))
@router.get("/lots/referentiel", response_model=list[LotReferentielResponse])
async def list_lots_referentiel(
immeuble_id: int | None = Query(None, description="Filtrer par immeuble"),
session: Session = Depends(get_session),
) -> list[LotReferentielResponse]:
"""Liste les lots avec leur fiche de caractéristiques.
- **immeuble_id**: ID de l'immeuble pour restreindre la liste (optionnel)
Tout le parc par défaut : le tableau de saisie porte une colonne immeuble,
et comparer deux immeubles au m² n'a de sens que s'ils s'affichent ensemble.
Les lots sans fiche sont renvoyés avec `caracteristiques` à `null` : le
tableau doit montrer les lignes vides autant que les remplies.
"""
if immeuble_id is not None and session.get(Immeuble, immeuble_id) is None:
raise HTTPException(status_code=404, detail="Immeuble introuvable.")
stmt = ( stmt = (
select(Lot, nb_revenus.label("nb_revenus"), nb_depenses.label("nb_depenses")) _requete_lignes()
.where(Lot.immeuble_id == immeuble_id) .add_columns(Immeuble.code, Immeuble.denomination)
.order_by(Lot.numero) .join(Immeuble, Lot.immeuble_id == Immeuble.id)
.order_by(Immeuble.code, Lot.numero)
) )
if immeuble_id is not None:
stmt = stmt.where(Lot.immeuble_id == immeuble_id)
return [ return [
_lot_response( _lot_response(
row.Lot, row.Lot,
immeuble.code, row.code,
immeuble_denomination=row.denomination,
nb_revenus=row.nb_revenus or 0, nb_revenus=row.nb_revenus or 0,
nb_depenses=row.nb_depenses or 0, nb_depenses=row.nb_depenses or 0,
) )
@@ -142,7 +160,89 @@ async def upsert_caracteristiques(
setattr(fiche, champ, getattr(body, champ)) setattr(fiche, champ, getattr(body, champ))
session.commit() session.commit()
session.refresh(lot)
row = session.execute(_requete_lignes().where(Lot.id == lot_id)).one()
immeuble = session.get(Immeuble, lot.immeuble_id) immeuble = session.get(Immeuble, lot.immeuble_id)
return _lot_response(lot, immeuble.code if immeuble else None) return _lot_response(
row.Lot,
immeuble.code if immeuble else None,
immeuble_denomination=immeuble.denomination if immeuble else None,
nb_revenus=row.nb_revenus or 0,
nb_depenses=row.nb_depenses or 0,
)
@router.delete("/lots/{lot_id}", status_code=204)
async def supprimer_lot(
lot_id: int,
session: Session = Depends(get_session),
) -> None:
"""Supprime un lot que rien ne rattache à un document.
- **lot_id**: ID du lot
Sert à nettoyer les lots laissés par un ancien format de numérotation, qui
encombrent le tableau de saisie sans rien décrire. Un lot qui porte des
revenus ou des dépenses est refusé : le supprimer emporterait des montants
du compte rendu, et un compte faux est pire qu'une ligne en trop.
"""
lot = session.get(Lot, lot_id)
if lot is None:
raise HTTPException(status_code=404, detail="Lot introuvable.")
row = session.execute(_requete_lignes().where(Lot.id == lot_id)).one()
if row.nb_revenus or row.nb_depenses:
raise HTTPException(
status_code=409,
detail=(
f"Le lot {lot.numero} porte {row.nb_revenus} revenu(s) et "
f"{row.nb_depenses} dépense(s) : il ne peut pas être supprimé."
),
)
# Les locataires du lot partent avec lui (cascade). Un locataire qui aurait
# encore des revenus les rattacherait au lot, deja refuse ci-dessus.
session.delete(lot)
session.commit()
@router.put("/immeubles/{immeuble_id}", response_model=ImmeubleResponse)
async def renommer_immeuble(
immeuble_id: int,
body: ImmeubleBody,
session: Session = Depends(get_session),
) -> ImmeubleResponse:
"""Donne à l'immeuble son nom d'usage.
- **immeuble_id**: ID de l'immeuble
Le code de gestion ("33689020") vient des PDF et ne se remplace pas ; la
dénomination ("Servient") s'affiche à sa place partout où l'immeuble est
cité.
"""
immeuble = session.get(Immeuble, immeuble_id)
if immeuble is None:
raise HTTPException(status_code=404, detail="Immeuble introuvable.")
immeuble.denomination = body.denomination
session.commit()
# Compteurs recalcules plutot que laisses a zero : la reponse remplace
# l'immeuble dans les listes du client, qui le croirait vide de lots.
nb_lots = session.execute(
select(func.count(Lot.id)).where(Lot.immeuble_id == immeuble.id)
).scalar_one()
nb_depenses = session.execute(
select(func.count(Depense.id)).where(Depense.immeuble_id == immeuble.id)
).scalar_one()
return ImmeubleResponse(
id=immeuble.id,
code=immeuble.code,
denomination=immeuble.denomination,
adresse=immeuble.adresse,
ville=immeuble.ville,
code_postal=immeuble.code_postal,
nb_lots=nb_lots,
nb_depenses=nb_depenses,
)

View File

@@ -8,7 +8,12 @@ PDF, la vider a moitie, ou la laisser contredire l'extraction en silence.
import pytest import pytest
from plesna_gerance.database.models import Depense, Immeuble, Lot from plesna_gerance.database.models import (
Depense,
Immeuble,
Lot,
LotCaracteristiques,
)
from plesna_gerance.database.service import DatabaseService from plesna_gerance.database.service import DatabaseService
@@ -41,7 +46,7 @@ def test_liste_les_lots_sans_fiche(api_client, immeuble_et_lot):
"""Un lot jamais decrit doit apparaitre, sinon il n'est pas saisissable.""" """Un lot jamais decrit doit apparaitre, sinon il n'est pas saisissable."""
immeuble, lot = immeuble_et_lot immeuble, lot = immeuble_et_lot
response = api_client.get(f"/api/immeubles/{immeuble.id}/lots/referentiel") response = api_client.get(f"/api/lots/referentiel?immeuble_id={immeuble.id}")
assert response.status_code == 200 assert response.status_code == 200
lignes = response.json() lignes = response.json()
@@ -55,7 +60,36 @@ def test_liste_les_lots_sans_fiche(api_client, immeuble_et_lot):
def test_liste_un_immeuble_inconnu_donne_404(api_client): def test_liste_un_immeuble_inconnu_donne_404(api_client):
assert api_client.get("/api/immeubles/99999/lots/referentiel").status_code == 404 assert api_client.get("/api/lots/referentiel?immeuble_id=99999").status_code == 404
def test_liste_tout_le_parc_par_defaut(api_client, immeuble_et_lot, db_session):
"""Le tableau porte une colonne immeuble : il les montre donc tous.
Restreindre a un immeuble par defaut obligerait a le choisir avant de voir
quoi que ce soit, alors que le parc entier tient dans un ecran.
"""
autre = Immeuble(code="IMM2", denomination="Marietton")
db_session.add(autre)
db_session.flush()
db_session.add(Lot(immeuble_id=autre.id, numero="01"))
db_session.commit()
lignes = api_client.get("/api/lots/referentiel").json()
assert len(lignes) == 2
assert {ligne["immeuble_code"] for ligne in lignes} == {"IMM1", "IMM2"}
def test_expose_le_nom_d_usage_de_l_immeuble(api_client, immeuble_et_lot):
"""Chaque ligne porte de quoi nommer son immeuble sans requete de plus."""
immeuble, _ = immeuble_et_lot
api_client.put(f"/api/immeubles/{immeuble.id}", json={"denomination": "Servient"})
ligne = api_client.get("/api/lots/referentiel").json()[0]
assert ligne["immeuble_denomination"] == "Servient"
assert ligne["immeuble_code"] == "IMM1"
def test_enregistre_puis_relit_une_fiche(api_client, immeuble_et_lot): def test_enregistre_puis_relit_une_fiche(api_client, immeuble_et_lot):
@@ -69,7 +103,7 @@ def test_enregistre_puis_relit_une_fiche(api_client, immeuble_et_lot):
assert fiche["etage"] == "RC" assert fiche["etage"] == "RC"
assert fiche["numero_fiscal"] == "690123456789" assert fiche["numero_fiscal"] == "690123456789"
relu = api_client.get(f"/api/immeubles/{immeuble.id}/lots/referentiel").json()[0] relu = api_client.get(f"/api/lots/referentiel?immeuble_id={immeuble.id}").json()[0]
assert relu["caracteristiques"]["surface"] == 148.0 assert relu["caracteristiques"]["surface"] == 148.0
assert relu["caracteristiques"]["dpe_classe"] == "C" assert relu["caracteristiques"]["dpe_classe"] == "C"
@@ -94,7 +128,7 @@ def test_une_seconde_ecriture_met_a_jour_sans_dupliquer(api_client, immeuble_et_
f"/api/lots/{lot.id}/caracteristiques", json={**FICHE, "surface": 150.0} f"/api/lots/{lot.id}/caracteristiques", json={**FICHE, "surface": 150.0}
) )
lignes = api_client.get(f"/api/immeubles/{immeuble.id}/lots/referentiel").json() lignes = api_client.get(f"/api/lots/referentiel?immeuble_id={immeuble.id}").json()
assert len(lignes) == 1 assert len(lignes) == 1
assert lignes[0]["caracteristiques"]["surface"] == 150.0 assert lignes[0]["caracteristiques"]["surface"] == 150.0
@@ -175,7 +209,7 @@ def test_une_re_extraction_ne_touche_pas_la_fiche(
DatabaseService(db_session).save_document(data=sample_data, overwrite=True) DatabaseService(db_session).save_document(data=sample_data, overwrite=True)
lignes = api_client.get(f"/api/immeubles/{immeuble.id}/lots/referentiel").json() lignes = api_client.get(f"/api/lots/referentiel?immeuble_id={immeuble.id}").json()
fiche = next(ligne for ligne in lignes if ligne["id"] == lot.id)["caracteristiques"] fiche = next(ligne for ligne in lignes if ligne["id"] == lot.id)["caracteristiques"]
assert fiche is not None, "la re-extraction a perdu la fiche du lot" assert fiche is not None, "la re-extraction a perdu la fiche du lot"
assert fiche["surface"] == 148.0 assert fiche["surface"] == 148.0
@@ -185,7 +219,7 @@ def test_compte_les_rattachements_du_lot(api_client, immeuble_et_lot):
"""Un lot avec des revenus n'est pas un orphelin : la liste doit le dire.""" """Un lot avec des revenus n'est pas un orphelin : la liste doit le dire."""
immeuble, lot = immeuble_et_lot immeuble, lot = immeuble_et_lot
ligne = api_client.get(f"/api/immeubles/{immeuble.id}/lots/referentiel").json()[0] ligne = api_client.get(f"/api/lots/referentiel?immeuble_id={immeuble.id}").json()[0]
assert ligne["nb_revenus"] == 1 assert ligne["nb_revenus"] == 1
# La depense de la fixture porte sur l'immeuble (lot_id NULL). La compter # La depense de la fixture porte sur l'immeuble (lot_id NULL). La compter
@@ -193,6 +227,21 @@ def test_compte_les_rattachements_du_lot(api_client, immeuble_et_lot):
assert ligne["nb_depenses"] == 0 assert ligne["nb_depenses"] == 0
def test_l_enregistrement_renvoie_les_memes_comptes_que_la_liste(
api_client, immeuble_et_lot
):
"""Décrire un lot occupé ne doit pas le faire passer pour inutilisé.
La réponse du PUT remplace la ligne dans le tableau : si elle rapporte zéro
revenu, le lot se pare d'un « inutilisé » que la liste dément au rechargement.
"""
_, lot = immeuble_et_lot
ligne = api_client.put(f"/api/lots/{lot.id}/caracteristiques", json=FICHE).json()
assert ligne["nb_revenus"] == 1
def test_compte_les_depenses_propres_au_lot(api_client, immeuble_et_lot, db_session): def test_compte_les_depenses_propres_au_lot(api_client, immeuble_et_lot, db_session):
"""Une depense rattachee au lot, elle, doit bien remonter sur sa ligne.""" """Une depense rattachee au lot, elle, doit bien remonter sur sa ligne."""
immeuble, lot = immeuble_et_lot immeuble, lot = immeuble_et_lot
@@ -200,6 +249,112 @@ def test_compte_les_depenses_propres_au_lot(api_client, immeuble_et_lot, db_sess
depense.lot_id = lot.id depense.lot_id = lot.id
db_session.commit() db_session.commit()
ligne = api_client.get(f"/api/immeubles/{immeuble.id}/lots/referentiel").json()[0] ligne = api_client.get(f"/api/lots/referentiel?immeuble_id={immeuble.id}").json()[0]
assert ligne["nb_depenses"] == 1 assert ligne["nb_depenses"] == 1
# ============================================================
# Suppression des lots sans rattachement
# ============================================================
def test_supprime_un_lot_inutilise(api_client, immeuble_et_lot, db_session):
"""Les lots d'un ancien format de numerotation doivent pouvoir disparaitre."""
immeuble, _ = immeuble_et_lot
orphelin = Lot(immeuble_id=immeuble.id, numero="0001", type="Appartement T1")
db_session.add(orphelin)
db_session.commit()
response = api_client.delete(f"/api/lots/{orphelin.id}")
assert response.status_code == 204
numeros = [
ligne["numero"]
for ligne in api_client.get(
f"/api/lots/referentiel?immeuble_id={immeuble.id}"
).json()
]
assert numeros == ["01"]
def test_supprimer_un_lot_emporte_sa_fiche(api_client, immeuble_et_lot, db_session):
"""Sans cascade, la fiche resterait en base sans lot pour la porter."""
immeuble, _ = immeuble_et_lot
orphelin = Lot(immeuble_id=immeuble.id, numero="0001")
db_session.add(orphelin)
db_session.commit()
api_client.put(f"/api/lots/{orphelin.id}/caracteristiques", json=FICHE)
api_client.delete(f"/api/lots/{orphelin.id}")
assert db_session.query(LotCaracteristiques).count() == 0
def test_refuse_de_supprimer_un_lot_avec_des_revenus(api_client, immeuble_et_lot):
"""Le refus protege des montants : les perdre fausserait les comptes."""
_, lot = immeuble_et_lot
response = api_client.delete(f"/api/lots/{lot.id}")
assert response.status_code == 409
assert "revenu" in response.json()["detail"]
def test_refuse_de_supprimer_un_lot_avec_des_depenses(
api_client, immeuble_et_lot, db_session
):
"""Une depense seule suffit a retenir le lot, meme sans aucun revenu."""
immeuble, _ = immeuble_et_lot
lot_charge = Lot(immeuble_id=immeuble.id, numero="0002")
db_session.add(lot_charge)
db_session.commit()
depense = db_session.query(Depense).one()
depense.lot_id = lot_charge.id
db_session.commit()
response = api_client.delete(f"/api/lots/{lot_charge.id}")
assert response.status_code == 409
def test_supprimer_un_lot_inconnu_donne_404(api_client):
assert api_client.delete("/api/lots/99999").status_code == 404
# ============================================================
# Nom d'usage de l'immeuble
# ============================================================
def test_nomme_un_immeuble(api_client, immeuble_et_lot):
immeuble, _ = immeuble_et_lot
response = api_client.put(
f"/api/immeubles/{immeuble.id}", json={"denomination": "Servient"}
)
assert response.status_code == 200
assert response.json()["denomination"] == "Servient"
assert api_client.get("/api/immeubles").json()[0]["denomination"] == "Servient"
# La reponse remplace l'immeuble dans les listes du client : des compteurs a
# zero le feraient passer pour un immeuble sans lot.
assert response.json()["nb_lots"] == 1
def test_effacer_le_nom_rend_l_immeuble_a_son_code(api_client, immeuble_et_lot):
immeuble, _ = immeuble_et_lot
api_client.put(f"/api/immeubles/{immeuble.id}", json={"denomination": "Servient"})
response = api_client.put(
f"/api/immeubles/{immeuble.id}", json={"denomination": " "}
)
assert response.json()["denomination"] is None
assert response.json()["code"] == "IMM1"
def test_nommer_un_immeuble_inconnu_donne_404(api_client):
response = api_client.put("/api/immeubles/99999", json={"denomination": "Servient"})
assert response.status_code == 404