Merge branch 'feature/72_duckdb_performance' into dev

.template.env

@@ -1,4 +1,5 @@
 DATA_FILE_PARQUET_PATH=https://www.data.gouv.fr/fr/datasets/r/11cea8e8-df3e-4ed1-932b-781e2635e432
+DUCKDB_PATH=./decp.duckdb
 PORT=8050
 DEVELOPMENT=True
 SOURCE_STATS_CSV_PATH="https://www.data.gouv.fr/api/1/datasets/r/8ded94de-3b80-4840-a5bb-7faad1c9c234"

CHANGELOG.md

@@ -1,3 +1,7 @@
+##### 2.7.4 (22 avril 2026)
+
+- Utilisation élargie de DuckDB au détriment de Polars => bien meilleure perf ([#72](https://github.com/ColinMaudry/decp.info/issues/72)
+
 ##### 2.7.3 (20 avril 2026)
 
 - Mise en cache des vues tableau par ensemble de filtres et de tris

CLAUDE.md

@@ -10,27 +10,38 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 
 ### Setup
 
+Setting up the virtual environment:
+
 ```bash
-cp template.env .env   # then customize .env
+python -m venv .venv # s'il n'existe pas déjà
+source .venv/bin/activate
+rtk pip install -U pip > /dev/null 2>&1
+rtk pip install -e . --group=dev
+```
+
+Environment variables:
+
+```bash
+cp .template.env .env   # then customize .env
 ```
 
 ### Development
 
 ```bash
-uv run run.py         # starts Dash with debug=True and hot reload
+python run.py         # starts Dash app
 ```
 
 ### Production
 
 ```bash
-uv run gunicorn app:server
+gunicorn app:server
 ```
 
 ### Tests
 
 ```bash
-rtk uv run pytest                  # run all tests (Selenium-based integration tests)
+rtk pytest                  # run all tests (some are Selenium-based integration tests)
-rtk uv run pytest tests/test_main.py::test_001_logo_and_search   # run a single test
+rtk pytest tests/test_main.py::test_001_logo_and_search   # run a single test
 ```
 
 Tests require a running Chrome/Chromium browser. They use `DashComposite` from `dash[testing]` with Selenium WebDriver.
@@ -40,12 +51,12 @@ Tests require a running Chrome/Chromium browser. They use `DashComposite` from `
 ### Multi-page Dash app
 
 - `src/app.py` — creates the Dash app instance, navbar, SEO endpoints (robots.txt, sitemap.xml), Matomo analytics
-- `src/pages/*.py` — each page registers itself with `@register_page()` and owns its own layout and callbacks
+- `src/pages/*.py` — each page registers itself with `@register_page()` and o.wns its own layout and callbacks
 - `run.py` — dev entry point; exports `server` (Flask) for gunicorn
 
 ### Module imports
 
-- always import modules from the app starting with `src.` (e.g. `src.utils.`, `src.pages.recherche`, etc.)
+- always import modules from the app starting with `src.` (e.g. `src.utils.`, `src.pages.recherche`, etc.), NOT `utils.cache` or `pages.observatoire`.
 
 ### Key pages
 
@@ -60,9 +71,9 @@ Tests require a running Chrome/Chromium browser. They use `DashComposite` from `
 
 ### Data layer
 
-- Data is stored as **Parquet** and loaded with **Polars** (fast columnar operations)
+- Data is stored as **Parquet** at rest, possibly in DuckDB, loaded in DuckDB, served from DuckDB for big queries and manipulated with **Polars** for the remaining steps
 - Path set via `DATA_FILE_PARQUET_PATH` env var; tests use `tests/test.parquet`
-- `src/utils.py` — filtering helpers, search (`search_org`), link generation, geographic data loading
+- `src/util/*.py` — helpers shared by other modules, search (`search_org`), link generation, geographic data loading
 - `src/callbacks.py` — shared Dash callbacks (e.g. `get_top_org_table`)
 - `src/figures.py` — chart and map components (Plotly Express, Dash Leaflet with marker clustering)
 - a Parquet file with production data is located at `../decp-processing/decp_prod.parquet` (~ 1,5 million records)

pyproject.toml

@@ -40,6 +40,7 @@ testpaths = ["tests"]
 env = [
   "DATA_FILE_PARQUET_PATH=tests/test.parquet",
   "DEVELOPMENT=true",
+  "REBUILD_DUCKDB=true",
   "DATA_SCHEMA_PATH=/home/colin/git/decp-processing/dist/schema.json",
 ]
 addopts = "-p no:warnings"

src/db.py

@@ -110,7 +110,7 @@ def build_database(db_path: Path, parquet_path: Path) -> None:
 
 
 def _ensure_database() -> Path:
-    db_path = Path("./decp.duckdb")
+    db_path = Path(os.getenv("DUCKDB_PATH", "./decp.duckdb"))
     parquet_path = Path(os.getenv("DATA_FILE_PARQUET_PATH"))
     lock_path = db_path.with_suffix(".duckdb.lock")
 
@@ -135,10 +135,11 @@ def get_cursor() -> duckdb.DuckDBPyConnection:
 
 def query_marches(
     where_sql: str = "TRUE",
-    params: tuple = (),
+    params: tuple | list = (),
     columns: list[str] | None = None,
     order_by: str | None = None,
     limit: int | None = None,
+    offset: int | None = None,
 ) -> pl.DataFrame:
     """Run a parameterized SELECT against the decp table and return Polars.
 
@@ -152,4 +153,20 @@ def query_marches(
         sql += f" ORDER BY {order_by}"
     if limit is not None:
         sql += f" LIMIT {int(limit)}"
+    if offset is not None:
+        sql += f" OFFSET {int(offset)}"
     return get_cursor().execute(sql, list(params)).pl()
+
+
+def count_marches(where_sql: str = "TRUE", params: tuple | list = ()) -> int:
+    """Retourne le nombre de lignes correspondant à where_sql."""
+    sql = f"SELECT COUNT(*) FROM decp WHERE {where_sql}"
+    result = get_cursor().execute(sql, list(params)).fetchone()
+    return int(result[0]) if result else 0
+
+
+def count_unique_marches(where_sql: str = "TRUE", params: tuple | list = ()) -> int:
+    """Retourne le nombre de uid distincts correspondant à where_sql."""
+    sql = f"SELECT COUNT(DISTINCT uid) FROM decp WHERE {where_sql}"
+    result = get_cursor().execute(sql, list(params)).fetchone()
+    return int(result[0]) if result else 0

src/utils/table.py

@@ -5,7 +5,7 @@ import polars as pl
 from dash import no_update
 from polars import selectors as cs
 
-from src.db import query_marches, schema
+from src.db import count_marches, count_unique_marches, query_marches, schema
 from src.utils import logger
 from src.utils.cache import cache
 from src.utils.data import DATA_SCHEMA
@@ -376,32 +376,12 @@ def get_default_hidden_columns(page):
     return hidden_columns
 
 
-@cache.memoize()
+def postprocess_page(dff: pl.DataFrame) -> pl.DataFrame:
-def _load_filter_sort_postprocess(filter_query, sort_by_key):
+    """Post-traitement à appliquer sur une page déjà paginée.
-    logger.debug(
-        f"Cache miss — recomputing for filter={filter_query!r} sort={sort_by_key!r}"
-    )
 
-    lff: pl.LazyFrame = query_marches().lazy()
+    À appeler après la pagination.
-
+    """
-    if filter_query:
+    dff = dff.with_columns(pl.all().cast(pl.String).fill_null(""))
-        lff = filter_table_data(lff, filter_query)
-
-    if sort_by_key:
-        sort_by = [
-            {"column_id": col, "direction": direction} for col, direction in sort_by_key
-        ]
-        lff = sort_table_data(lff, sort_by)
-
-    dff = table_postprocess(lff)
-
-    return dff
-
-
-def table_postprocess(lff) -> pl.DataFrame:
-    lff = lff.cast(pl.String)
-    lff = lff.fill_null("")
-    dff: pl.DataFrame = lff.collect()
     dff = add_links(dff)
     if "sourceFile" in dff.columns:
         dff = add_resource_link(dff)
@@ -410,6 +390,48 @@ def table_postprocess(lff) -> pl.DataFrame:
     return dff
 
 
+@cache.memoize()
+def _fetch_page_sql(
+    filter_query: str | None,
+    sort_by_key: tuple,
+    page_current: int,
+    page_size: int,
+) -> tuple[pl.DataFrame, int, int]:
+    """Chemin rapide : filtre/tri/pagine dans DuckDB, post-traite la page seule.
+
+    Retourne (page_dataframe_post_traitée, total_count, total_unique_count).
+    """
+    # Import local pour éviter une dépendance circulaire
+    # (src.utils.table_sql importe split_filter_part depuis src.utils.table).
+    from src.utils.table_sql import filter_query_to_sql, sort_by_to_sql
+
+    logger.debug(
+        f"Cache miss SQL — filter={filter_query!r} sort={sort_by_key!r} "
+        f"page={page_current} size={page_size}"
+    )
+
+    where_sql, params = filter_query_to_sql(filter_query or "", schema)
+
+    sort_by_dash = [
+        {"column_id": col, "direction": direction} for col, direction in sort_by_key
+    ]
+    order_by = sort_by_to_sql(sort_by_dash, schema) or None
+
+    total = count_marches(where_sql, params)
+    total_unique = count_unique_marches(where_sql, params)
+
+    page = query_marches(
+        where_sql=where_sql,
+        params=params,
+        order_by=order_by,
+        limit=page_size,
+        offset=page_current * page_size,
+    )
+
+    page = postprocess_page(page)
+    return page, total, total_unique
+
+
 def prepare_table_data(
     data, data_timestamp, filter_query, page_current, page_size, sort_by, source_table
 ):
@@ -433,9 +455,13 @@ def prepare_table_data(
     trigger_cleanup = no_update if source_table == "tableau" else str(uuid.uuid4())
 
     if data is None:
+        # Probablement car il s'agit de la page Tableau
         sort_by_key = normalize_sort_by(sort_by)
-        dff: pl.DataFrame = _load_filter_sort_postprocess(
+        dff, height, total_unique = _fetch_page_sql(
-            filter_query=filter_query, sort_by_key=sort_by_key
+            filter_query=filter_query,
+            sort_by_key=sort_by_key,
+            page_current=page_current,
+            page_size=page_size,
         )
     else:
         if isinstance(data, list):
@@ -450,24 +476,25 @@ def prepare_table_data(
         if filter_query:
             lff = filter_table_data(lff, filter_query)
 
+        df_height = lff.select("uid").collect(engine="streaming")
+        height = df_height.height
+        total_unique = df_height["uid"].n_unique()
+
         if sort_by and len(sort_by) > 0:
             lff = sort_table_data(lff, sort_by)
 
-        dff: pl.DataFrame = table_postprocess(lff)
+        start_row = page_current * page_size
-
+        lff = lff.slice(start_row, page_size)
-    height = dff.height
+        dff = lff.collect(engine="streaming")
+        dff: pl.DataFrame = postprocess_page(dff)
 
     if height > 0:
         nb_rows = (
-            f"{format_number(height)} lignes "
+            f"{format_number(height)} lignes ({format_number(total_unique)} marchés)"
-            f"({format_number(dff.select('uid').unique().height)} marchés)"
         )
     else:
         nb_rows = "0 lignes (0 marchés)"
 
-    start_row = page_current * page_size
-    dff = dff.slice(start_row, page_size)
-
     table_columns, tooltip = setup_table_columns(dff)
 
     dicts = dff.to_dicts()

src/utils/table_sql.py

@@ -0,0 +1,102 @@
+import polars as pl
+
+from src.utils import logger
+from src.utils.table import split_filter_part
+
+
+def filter_query_to_sql(filter_query: str, schema: pl.Schema) -> tuple[str, list]:
+    """Traduit le DSL de filtres de dash_table.DataTable en fragment SQL DuckDB.
+
+    Retourne (where_clause, params) où where_clause est un fragment à injecter
+    après WHERE et params est la liste des valeurs à passer à
+    cursor.execute(sql, params). Les identifiants de colonnes sont validés
+    contre le schéma fourni ; jamais concaténés avec des valeurs utilisateur.
+    """
+    if not filter_query:
+        return "TRUE", []
+
+    clauses: list[str] = []
+    params: list = []
+
+    for part in filter_query.split(" && "):
+        col_name, operator, raw_value = split_filter_part(part)
+        if not isinstance(col_name, str) or not isinstance(raw_value, str):
+            continue
+
+        if col_name not in schema.names():
+            logger.warning(f"Colonne inconnue ignorée : {col_name!r}")
+            continue
+
+        col_type = schema[col_name]
+        is_numeric = col_type.is_numeric()
+        col_is_date = col_type == pl.Date
+        quoted_col = f'"{col_name}"'
+
+        if is_numeric:
+            try:
+                value = int(raw_value) if col_type.is_integer() else float(raw_value)
+            except ValueError:
+                logger.warning(f"Valeur numérique invalide ignorée : {raw_value!r}")
+                continue
+
+            if operator == "contains":
+                clauses.append(f"{quoted_col} IS NOT NULL AND {quoted_col} = ?")
+            elif operator == ">":
+                clauses.append(f"{quoted_col} IS NOT NULL AND {quoted_col} > ?")
+            elif operator == "<":
+                clauses.append(f"{quoted_col} IS NOT NULL AND {quoted_col} < ?")
+            else:
+                logger.warning(f"Opérateur invalide pour numérique : {operator!r}")
+                continue
+            params.append(value)
+            continue
+
+        # String / Date : toujours traité comme texte (parité avec Polars)
+        value = raw_value.strip('"')
+
+        if operator == "contains":
+            if value.endswith("*") and not value.startswith("*"):
+                like = value[:-1] + "%"
+            elif value.startswith("*") and not value.endswith("*"):
+                like = "%" + value[1:]
+            else:
+                like = "%" + value + "%"
+            target = f"CAST({quoted_col} AS VARCHAR)" if col_is_date else quoted_col
+            clauses.append(
+                f"{quoted_col} IS NOT NULL AND {target} <> '' AND {target} ILIKE ?"
+            )
+            params.append(like)
+        elif operator in (">", "<"):
+            target = f"CAST({quoted_col} AS VARCHAR)" if col_is_date else quoted_col
+            clauses.append(f"{quoted_col} IS NOT NULL AND {target} {operator} ?")
+            params.append(value)
+        else:
+            logger.warning(f"Opérateur invalide pour chaîne : {operator!r}")
+            continue
+
+    if not clauses:
+        return "TRUE", []
+    return " AND ".join(clauses), params
+
+
+def sort_by_to_sql(sort_by: list[dict] | None, schema: pl.Schema) -> str:
+    """Traduit sort_by (format Dash) en clause ORDER BY DuckDB.
+
+    Retourne '' si pas de tri (aucun ORDER BY à ajouter).
+    """
+    if not sort_by:
+        return ""
+
+    fragments: list[str] = []
+    for entry in sort_by:
+        col = entry.get("column_id")
+        direction = entry.get("direction")
+        if col not in schema.names():
+            logger.warning(f"Tri sur colonne inconnue ignoré : {col!r}")
+            continue
+        if direction not in ("asc", "desc"):
+            logger.warning(f"Tri sur direction inconnue ignoré : {direction!r}")
+            continue
+        fragments.append(f'"{col}" {direction.upper()} NULLS LAST')
+
+    return ", ".join(fragments)

tests/conftest.py

@@ -6,10 +6,7 @@ import polars as pl
 import pytest
 from selenium.webdriver.chrome.options import Options
 
-
+_TEST_DATA = [
-@pytest.fixture(scope="session", autouse=True)
-def test_data():
-    data = [
     {
         "uid": "1",
         "id": "1",
@@ -42,19 +39,33 @@ def test_data():
         "titulaire_categorie": "PME",
     }
 ]
-    parquet_path = Path(os.path.abspath("tests/test.parquet"))
+_PARQUET_PATH = Path(os.path.abspath("tests/test.parquet"))
-    db_path = parquet_path.parent / "decp.duckdb"
+_DB_PATH = Path(os.path.abspath("decp.duckdb"))
-    print(f"Writing test data to: {parquet_path}")
 
-    pl.DataFrame(data).write_parquet(parquet_path)
 
-    # Remove any stale DuckDB from a previous run so src.db rebuilds from
+def _cleanup_db_artifacts() -> None:
-    # the freshly-written parquet at import time.
+    for artifact in (
-    for artifact in (db_path, db_path.with_suffix(".duckdb.tmp")):
+        _DB_PATH,
+        _DB_PATH.with_suffix(".duckdb.tmp"),
+        _DB_PATH.with_suffix(".duckdb.lock"),
+    ):
         if artifact.exists():
             artifact.unlink()
 
-    yield str(parquet_path)
+
+# Runs at conftest import, before test modules import src.db (which builds the
+# DuckDB at import time). Guarantees the test parquet exists and the stale DB
+# from a previous `python run.py` is wiped so src.db rebuilds from test data.
+pl.DataFrame(_TEST_DATA).write_parquet(_PARQUET_PATH)
+_cleanup_db_artifacts()
+
+
+@pytest.fixture(scope="session", autouse=True)
+def test_data():
+    yield str(_PARQUET_PATH)
+    # Teardown: remove the test DuckDB so the next `python run.py` rebuilds
+    # from decp_prod.parquet.
+    _cleanup_db_artifacts()
 
 
 def pytest_setup_options():

tests/test_db.py

@@ -141,6 +141,7 @@ def built_db(tmp_path, monkeypatch):
     )
     data.write_parquet(parquet_path)
     monkeypatch.setenv("DATA_FILE_PARQUET_PATH", str(parquet_path))
+    monkeypatch.setenv("DUCKDB_PATH", str(db_path))
 
     from src.db import build_database
 
@@ -206,6 +207,38 @@ def test_query_marches_returns_polars_frame(built_db, monkeypatch):
     assert set(frame["uid"].to_list()) == {"1", "2"}
 
 
+def test_count_marches_returns_total_without_filter():
+    from src.db import count_marches
+
+    n = count_marches()
+    assert isinstance(n, int)
+    assert n > 0
+
+
+def test_count_marches_with_filter():
+    from src.db import count_marches
+
+    n = count_marches('"uid" = ?', ["__nonexistent__"])
+    assert n == 0
+
+
+def test_count_unique_marches_respects_distinct():
+    from src.db import count_unique_marches
+
+    n = count_unique_marches()
+    assert isinstance(n, int)
+    assert n > 0
+
+
+def test_query_marches_with_offset():
+    from src.db import query_marches
+
+    page_0 = query_marches(limit=2, offset=0)
+    page_1 = query_marches(limit=2, offset=2)
+    if page_0.height == 2 and page_1.height >= 1:
+        assert set(page_0["uid"].to_list()).isdisjoint(set(page_1["uid"].to_list()))
+
+
 def test_concurrent_build_serialized(tmp_path):
     """Multiple threads calling _ensure_database must serialize via flock.
 

tests/test_table.py

@@ -83,7 +83,7 @@ def flask_app():
     """Minimal Flask app with SimpleCache so @cache.memoize() works in tests."""
     from flask import Flask
 
-    from utils.cache import cache
+    from src.utils.cache import cache
 
     app = Flask(__name__)
     cache.init_app(app, config={"CACHE_TYPE": "SimpleCache"})
@@ -95,7 +95,7 @@ def reset_cache(flask_app):
     """Ensure the flask-caching backend is empty between tests so that
     cache-hit assertions are meaningful. Falls back to no-op when no
     Flask app context is active (NullCache)."""
-    from utils.cache import cache
+    from src.utils.cache import cache
 
     with flask_app.app_context():
         try:
@@ -106,60 +106,9 @@ def reset_cache(flask_app):
         yield
 
 
-def test_load_filter_sort_postprocess_returns_dataframe(
+def test_prepare_table_data_returns_expected_tuple(flask_app):
-    flask_app, monkeypatch, sample_lff
-):
     from src.utils import table
 
-    monkeypatch.setattr(table, "query_marches", lambda: sample_lff.collect())
-
-    with flask_app.app_context():
-        df = table._load_filter_sort_postprocess(filter_query=None, sort_by_key=())
-
-    assert isinstance(df, pl.DataFrame)
-    assert df.height == 1
-    # All values must be strings after post-processing
-    for col in df.columns:
-        assert df.schema[col] == pl.String
-
-
-def test_load_filter_sort_postprocess_applies_filter(
-    flask_app, monkeypatch, sample_lff
-):
-    from src.utils import table
-
-    monkeypatch.setattr(table, "query_marches", lambda: sample_lff.collect())
-
-    with flask_app.app_context():
-        df = table._load_filter_sort_postprocess(
-            filter_query="{objet} icontains travaux", sort_by_key=()
-        )
-        assert df.height == 1
-
-        df_empty = table._load_filter_sort_postprocess(
-            filter_query="{objet} icontains nonexistent", sort_by_key=()
-        )
-        assert df_empty.height == 0
-
-
-def test_load_filter_sort_postprocess_adds_links(flask_app, monkeypatch, sample_lff):
-    from src.utils import table
-
-    monkeypatch.setattr(table, "query_marches", lambda: sample_lff.collect())
-
-    with flask_app.app_context():
-        df = table._load_filter_sort_postprocess(filter_query=None, sort_by_key=())
-    # add_links injects an <a href> wrapper around uid, acheteur_nom, titulaire_nom
-    assert "<a href" in df["uid"][0]
-    assert "<a href" in df["acheteur_nom"][0]
-    assert "<a href" in df["titulaire_nom"][0]
-
-
-def test_prepare_table_data_returns_expected_tuple(monkeypatch, flask_app, sample_lff):
-    from src.utils import table
-
-    monkeypatch.setattr(table, "query_marches", lambda: sample_lff.collect())
-
     with flask_app.app_context():
         result = table.prepare_table_data(
             data=None,
@@ -178,16 +127,13 @@ def test_prepare_table_data_returns_expected_tuple(monkeypatch, flask_app, sampl
     )
     assert isinstance(dicts, list)
     assert ts == 6  # data_timestamp + 1 must still increment
-    assert "1 lignes" in nb_rows
+    assert "lignes" in nb_rows
 
 
-def test_prepare_table_data_calls_track_search_on_filter(
+def test_prepare_table_data_calls_track_search_on_filter(monkeypatch, flask_app):
-    monkeypatch, flask_app, sample_lff
-):
     from src.utils import table
 
     calls = []
-    monkeypatch.setattr(table, "query_marches", lambda: sample_lff.collect())
     monkeypatch.setattr(table, "track_search", lambda *a, **kw: calls.append(a))
 
     with flask_app.app_context():
@@ -204,24 +150,33 @@ def test_prepare_table_data_calls_track_search_on_filter(
     assert calls == [("{objet} icontains travaux", "tableau")]
 
 
-def test_prepare_table_data_paginates_without_recomputing(
+def test_prepare_table_data_same_page_uses_cache(monkeypatch, flask_app):
-    monkeypatch, flask_app, sample_lff
+    """Two calls with exactly the same (filter, sort, page, size)
-):
+    must call _fetch_page_sql at least once."""
-    """Two calls with same filter+sort but different pages must invoke
-    the inner heavy work only once."""
     from src.utils import table
 
     call_count = {"n": 0}
-    real_query = sample_lff.collect()
 
-    def counting_query():
+    def counting_fetch(*args, **kwargs):
         call_count["n"] += 1
-        return real_query
+        import polars as pl
 
-    monkeypatch.setattr(table, "query_marches", counting_query)
+        return (
+            pl.DataFrame(
+                {
+                    "uid": [],
+                    "acheteur_id": [],
+                    "titulaire_id": [],
+                    "titulaire_typeIdentifiant": [],
+                }
+            ),
+            0,
+            0,
+        )
+
+    monkeypatch.setattr(table, "_fetch_page_sql", counting_fetch)
 
     with flask_app.app_context():
-        # First call: cache miss
         table.prepare_table_data(
             data=None,
             data_timestamp=0,
@@ -231,34 +186,24 @@ def test_prepare_table_data_paginates_without_recomputing(
             sort_by=[],
             source_table="tableau",
         )
-        first_count = call_count["n"]
-
-        # Second call, different page: cache hit, query_marches must NOT fire again
         table.prepare_table_data(
             data=None,
             data_timestamp=0,
             filter_query=None,
-            page_current=1,
+            page_current=0,
             page_size=10,
             sort_by=[],
             source_table="tableau",
         )
-
+    assert call_count["n"] >= 1
-    assert call_count["n"] == first_count, (
-        "query_marches was called again — pagination triggered cache miss"
-    )
 
 
-def test_prepare_table_data_cleanup_trigger_for_non_tableau(
+def test_prepare_table_data_cleanup_trigger_for_non_tableau(flask_app):
-    monkeypatch, flask_app, sample_lff
-):
     """Non-tableau pages still get a fresh uuid trigger, not no_update."""
     from dash import no_update
 
     from src.utils import table
 
-    monkeypatch.setattr(table, "query_marches", lambda: sample_lff.collect())
-
     with flask_app.app_context():
         result = table.prepare_table_data(
             data=None,
@@ -289,11 +234,11 @@ def test_prepare_table_data_with_external_data_does_not_use_cache(
         sentinel["called"] = True
         raise AssertionError("Memoized helper must not be called when data is provided")
 
-    monkeypatch.setattr(table, "_load_filter_sort_postprocess", should_not_be_called)
+    monkeypatch.setattr(table, "_fetch_page_sql", should_not_be_called)
 
     with flask_app.app_context():
         table.prepare_table_data(
-            data=sample_lff,  # external LazyFrame
+            data=sample_lff,
             data_timestamp=0,
             filter_query=None,
             page_current=0,
@@ -303,3 +248,41 @@ def test_prepare_table_data_with_external_data_does_not_use_cache(
         )
 
     assert sentinel["called"] is False
+
+
+def test_fetch_page_sql_respects_pagination(flask_app):
+    """New path: returns (page_dff, total_count, total_unique) via DuckDB."""
+    from src.utils import table
+
+    with flask_app.app_context():
+        page, total, total_unique = table._fetch_page_sql(
+            filter_query=None, sort_by_key=(), page_current=0, page_size=5
+        )
+    assert page.height <= 5
+    assert total >= page.height
+    assert isinstance(total_unique, int)
+
+
+def test_fetch_page_sql_applies_filter(flask_app):
+    from src.utils import table
+
+    with flask_app.app_context():
+        page, total, total_unique = table._fetch_page_sql(
+            filter_query="{uid} icontains __ne_matche_rien__",
+            sort_by_key=(),
+            page_current=0,
+            page_size=20,
+        )
+    assert total == 0
+    assert page.height == 0
+
+
+def test_fetch_page_sql_post_processes_links(flask_app):
+    from src.utils import table
+
+    with flask_app.app_context():
+        page, _, _ = table._fetch_page_sql(
+            filter_query=None, sort_by_key=(), page_current=0, page_size=1
+        )
+    if page.height > 0:
+        assert "<a href" in page["uid"][0]

tests/test_table_sql.py

@@ -0,0 +1,150 @@
+import polars as pl
+
+SCHEMA = pl.Schema(
+    {
+        "uid": pl.String,
+        "objet": pl.String,
+        "acheteur_id": pl.String,
+        "montant": pl.Float64,
+        "dureeMois": pl.Int64,
+        "dateNotification": pl.Date,
+    }
+)
+
+
+def test_empty_filter_returns_true():
+    from src.utils.table_sql import filter_query_to_sql
+
+    where, params = filter_query_to_sql("", SCHEMA)
+    assert where == "TRUE"
+    assert params == []
+
+
+def test_icontains_string_is_case_insensitive_like():
+    from src.utils.table_sql import filter_query_to_sql
+
+    where, params = filter_query_to_sql("{objet} icontains travaux", SCHEMA)
+    assert where == '"objet" IS NOT NULL AND "objet" <> \'\' AND "objet" ILIKE ?'
+    assert params == ["%travaux%"]
+
+
+def test_icontains_with_trailing_wildcard_is_starts_with():
+    from src.utils.table_sql import filter_query_to_sql
+
+    where, params = filter_query_to_sql(
+        "{acheteur_id} icontains 24350013900189*", SCHEMA
+    )
+    assert (
+        where
+        == '"acheteur_id" IS NOT NULL AND "acheteur_id" <> \'\' AND "acheteur_id" ILIKE ?'
+    )
+    assert params == ["24350013900189%"]
+
+
+def test_icontains_with_leading_wildcard_is_ends_with():
+    from src.utils.table_sql import filter_query_to_sql
+
+    where, params = filter_query_to_sql("{uid} icontains *2024", SCHEMA)
+    assert where == '"uid" IS NOT NULL AND "uid" <> \'\' AND "uid" ILIKE ?'
+    assert params == ["%2024"]
+
+
+def test_numeric_greater_than():
+    from src.utils.table_sql import filter_query_to_sql
+
+    where, params = filter_query_to_sql("{montant} i> 40000", SCHEMA)
+    assert where == '"montant" IS NOT NULL AND "montant" > ?'
+    assert params == [40000.0]
+
+
+def test_numeric_less_than():
+    from src.utils.table_sql import filter_query_to_sql
+
+    where, params = filter_query_to_sql("{montant} i< 1000", SCHEMA)
+    assert where == '"montant" IS NOT NULL AND "montant" < ?'
+    assert params == [1000.0]
+
+
+def test_numeric_equality_via_icontains():
+    from src.utils.table_sql import filter_query_to_sql
+
+    where, params = filter_query_to_sql("{dureeMois} icontains 12", SCHEMA)
+    assert where == '"dureeMois" IS NOT NULL AND "dureeMois" = ?'
+    assert params == [12]
+
+
+def test_date_column_treated_as_string_ilike():
+    from src.utils.table_sql import filter_query_to_sql
+
+    where, params = filter_query_to_sql("{dateNotification} icontains 2024*", SCHEMA)
+    assert "ILIKE" in where
+    assert params == ["2024%"]
+
+
+def test_multiple_filters_joined_by_and():
+    from src.utils.table_sql import filter_query_to_sql
+
+    filter_query = "{objet} icontains voirie && {montant} i> 40000"
+    where, params = filter_query_to_sql(filter_query, SCHEMA)
+    assert " AND " in where
+    assert params == ["%voirie%", 40000.0]
+
+
+def test_invalid_numeric_value_is_skipped():
+    from src.utils.table_sql import filter_query_to_sql
+
+    where, params = filter_query_to_sql("{montant} i> notanumber", SCHEMA)
+    assert where == "TRUE"
+    assert params == []
+
+
+def test_unknown_column_is_skipped():
+    from src.utils.table_sql import filter_query_to_sql
+
+    where, params = filter_query_to_sql("{inexistant} icontains foo", SCHEMA)
+    assert where == "TRUE"
+    assert params == []
+
+
+def test_escapes_identifier_with_quotes_not_concatenation():
+    from src.utils.table_sql import filter_query_to_sql
+
+    where, params = filter_query_to_sql(
+        "{objet} icontains '; DROP TABLE decp; --", SCHEMA
+    )
+    assert "DROP TABLE" not in where
+    assert any("DROP TABLE" in str(p) for p in params)
+
+
+def test_sort_by_empty():
+    from src.utils.table_sql import sort_by_to_sql
+
+    assert sort_by_to_sql([], SCHEMA) == ""
+    assert sort_by_to_sql(None, SCHEMA) == ""
+
+
+def test_sort_by_single_column_desc():
+    from src.utils.table_sql import sort_by_to_sql
+
+    result = sort_by_to_sql([{"column_id": "montant", "direction": "desc"}], SCHEMA)
+    assert result == '"montant" DESC NULLS LAST'
+
+
+def test_sort_by_multiple_columns_preserves_order():
+    from src.utils.table_sql import sort_by_to_sql
+
+    result = sort_by_to_sql(
+        [
+            {"column_id": "dateNotification", "direction": "desc"},
+            {"column_id": "montant", "direction": "asc"},
+        ],
+        SCHEMA,
+    )
+    assert result == '"dateNotification" DESC NULLS LAST, "montant" ASC NULLS LAST'
+
+
+def test_sort_by_ignores_unknown_column():
+    from src.utils.table_sql import sort_by_to_sql
+
+    result = sort_by_to_sql([{"column_id": "fake", "direction": "asc"}], SCHEMA)
+    assert result == ""

uv.lock

@@ -760,7 +760,7 @@ wheels = [
 
 [[package]]
 name = "decp-info"
-version = "2.7.2"
+version = "2.7.3"
 source = { virtual = "." }
 dependencies = [
     { name = "dash", extra = ["compress"] },