feat(entity_grid): fonctions pures de grille scopée acheteur/titulaire (#41)

2026-07-13 09:58:57 +02:00
parent 94addf7fae
commit 76c1096f4c
2 changed files with 225 additions and 0 deletions
@@ -0,0 +1,118 @@
+"""Logique de grille AG Grid scopée à une entité (acheteur/titulaire).
+
+Factorise ce qui est commun aux pages acheteur.py et titulaire.py :
+- scope SQL (WHERE acheteur_id/titulaire_id + année),
+- datasource server-side (réutilise fetch_grid_page du Lot 1),
+- export Excel filtré (réutilise export_dataframe),
+- columnDefs + réapplication de la disposition persistée,
+- fabrique de grille à id pattern-matching (persistance filterModel par fiche).
+
+Les callbacks Dash sont enregistrés par register_entity_grid_callbacks()
+(cf. module séparé de wiring) ; ici, seules des fonctions pures testables.
+"""
+
+from dash import no_update
+
+from src.figures import (
+    AG_GRID_LOCALE_FR,  # noqa: F401  (réexport pratique)
+    ag_grid,
+)
+from src.utils.grid import (
+    apply_persisted_layout,
+    export_dataframe,
+    fetch_grid_page,
+    grid_column_defs,
+)
+
+_TOUTES = "Toutes les années"
+
+
+def grid_type(org_type: str) -> str:
+    """Valeur de la clé `type` de l'id pattern-matching de la grille."""
+    return f"{org_type}-grid"
+
+
+def entity_scope(org_type: str, entity_id: str, year) -> tuple[str, list]:
+    """WHERE SQL + params liés scopant les requêtes à cette entité (+ année).
+
+    Reproduit _acheteur_scope/_titulaire_scope des pages, unifié par org_type.
+    """
+    if org_type == "titulaire":
+        where_sql = "titulaire_id = ? AND titulaire_typeIdentifiant = 'SIRET'"
+    else:
+        where_sql = "acheteur_id = ?"
+    params: list = [entity_id]
+    if year and year != _TOUTES:
+        where_sql += ' AND YEAR("dateNotification") = ?'
+        params.append(int(year))
+    return where_sql, params
+
+
+def _sort_model_from_column_state(column_state) -> list:
+    """Extrait le sortModel AG Grid d'un columnState (comme tableau.download_data)."""
+    return [
+        {"colId": c["colId"], "sort": c["sort"]}
+        for c in (column_state or [])
+        if c.get("sort")
+    ]
+
+
+def entity_grid_column_defs(hidden_columns, column_state) -> list[dict]:
+    """columnDefs du schéma DECP + disposition (largeur/ordre) persistée."""
+    defs = grid_column_defs(hidden_columns)
+    return apply_persisted_layout(defs, column_state)
+
+
+def fetch_entity_page(org_type, entity_id, year, request) -> tuple:
+    """Datasource server-side scopé pour la grille entité.
+
+    Renvoie ({"rowData": ..., "rowCount": total}, total, total_unique).
+    """
+    if request is None:
+        return no_update, no_update, no_update
+    base_where_sql, base_params = entity_scope(org_type, entity_id, year)
+    rows, total, total_unique = fetch_grid_page(
+        request.get("filterModel") or None,
+        request.get("sortModel") or None,
+        request.get("startRow", 0),
+        request.get("endRow", 100),
+        base_where_sql=base_where_sql,
+        base_params=tuple(base_params),
+    )
+    return {"rowData": rows, "rowCount": total}, total, total_unique
+
+
+def export_entity_dataframe(org_type, entity_id, year, filter_model, column_state):
+    """DataFrame filtré/trié (état courant de la grille) pour l'export Excel."""
+    base_where_sql, base_params = entity_scope(org_type, entity_id, year)
+    sort_model = _sort_model_from_column_state(column_state)
+    hidden_columns = [c["colId"] for c in (column_state or []) if c.get("hide")]
+    return export_dataframe(
+        filter_model,
+        sort_model,
+        hidden_columns,
+        base_where_sql=base_where_sql,
+        base_params=tuple(base_params),
+    )
+
+
+def clear_sort(column_state) -> list[dict]:
+    """columnState avec le tri effacé (sort/sortIndex à None), largeur/ordre/
+    épinglage préservés. Approche #47-safe de tableau.reset_view."""
+    return [{**col, "sort": None, "sortIndex": None} for col in (column_state or [])]
+
+
+def build_entity_grid(org_type, entity_id, year, hidden_columns, column_state):
+    """Grille AG Grid à id pattern-matching, scopée à (entité, année).
+
+    L'id inclut entity_id + year → une entrée localStorage de filterModel par
+    fiche/année (persistance native scopée). columnState n'est PAS persisté
+    nativement (géré par un store global partagé) : persisted_props=["filterModel"].
+    """
+    grid_id = {
+        "type": grid_type(org_type),
+        "entity_id": entity_id,
+        "year": year or _TOUTES,
+    }
+    defs = entity_grid_column_defs(hidden_columns, column_state)
+    return ag_grid(grid_id, defs, persisted_props=["filterModel"])
@@ -0,0 +1,107 @@
+import dash_ag_grid as dag
+
+import src.app  # noqa: F401  # instancie l'app → schema chargé
+from src.db import query_marches
+from src.utils.entity_grid import (
+    build_entity_grid,
+    clear_sort,
+    entity_grid_column_defs,
+    entity_scope,
+    export_entity_dataframe,
+    fetch_entity_page,
+    grid_type,
+)
+
+
+def _an_acheteur_id():
+    return query_marches("TRUE", (), columns=["acheteur_id"])["acheteur_id"][0]
+
+
+def test_entity_scope_acheteur():
+    where, params = entity_scope("acheteur", "S123", None)
+    assert where == "acheteur_id = ?"
+    assert params == ["S123"]
+
+
+def test_entity_scope_titulaire_requires_siret():
+    where, params = entity_scope("titulaire", "S123", None)
+    assert "titulaire_id = ?" in where
+    assert "titulaire_typeIdentifiant = 'SIRET'" in where
+    assert params == ["S123"]
+
+
+def test_entity_scope_with_year_adds_bound_param():
+    where, params = entity_scope("acheteur", "S123", "2025")
+    assert 'YEAR("dateNotification") = ?' in where
+    assert params == ["S123", 2025]
+
+
+def test_entity_scope_toutes_les_annees_no_year_filter():
+    where, params = entity_scope("acheteur", "S123", "Toutes les années")
+    assert "YEAR" not in where
+    assert params == ["S123"]
+
+
+def test_fetch_entity_page_is_scoped():
+    ach = _an_acheteur_id()
+    request = {"startRow": 0, "endRow": 50, "filterModel": None, "sortModel": None}
+    response, total, total_unique = fetch_entity_page("acheteur", ach, None, request)
+    assert total > 0
+    assert total_unique > 0
+    assert response["rowCount"] == total
+    assert len(response["rowData"]) <= 50
+    # Toutes les lignes appartiennent à l'acheteur scopé.
+    assert all(str(r["acheteur_id"]).find(str(ach)) != -1 for r in response["rowData"])
+
+
+def test_fetch_entity_page_none_request():
+    from dash import no_update
+
+    out = fetch_entity_page("acheteur", "S123", None, None)
+    assert out == (no_update, no_update, no_update)
+
+
+def test_export_entity_dataframe_scoped():
+    ach = _an_acheteur_id()
+    df = export_entity_dataframe("acheteur", ach, None, None, None)
+    assert df.height > 0
+    assert set(df["acheteur_id"].to_list()) == {ach}
+
+
+def test_clear_sort_preserves_width_and_order():
+    column_state = [
+        {"colId": "objet", "width": 400, "sort": "asc", "sortIndex": 0},
+        {"colId": "montant", "width": 150, "sort": "desc", "sortIndex": 1},
+    ]
+    cleared = clear_sort(column_state)
+    assert all(c["sort"] is None and c["sortIndex"] is None for c in cleared)
+    assert cleared[0]["width"] == 400 and cleared[0]["colId"] == "objet"
+    assert cleared[1]["width"] == 150
+
+
+def test_clear_sort_empty():
+    assert clear_sort(None) == []
+
+
+def test_entity_grid_column_defs_applies_persisted_layout():
+    defs = entity_grid_column_defs(
+        hidden_columns=[], column_state=[{"colId": "montant", "width": 999}]
+    )
+    by_field = {d["field"]: d for d in defs}
+    assert by_field["montant"]["width"] == 999
+
+
+def test_build_entity_grid_has_pattern_matching_id_and_filter_only_persistence():
+    grid = build_entity_grid("acheteur", "S123", "Toutes les années", [], None)
+    assert isinstance(grid, dag.AgGrid)
+    assert grid.id == {
+        "type": "acheteur-grid",
+        "entity_id": "S123",
+        "year": "Toutes les années",
+    }
+    assert list(grid.persisted_props) == ["filterModel"]
+
+
+def test_grid_type():
+    assert grid_type("acheteur") == "acheteur-grid"
+    assert grid_type("titulaire") == "titulaire-grid"