colibre/docs/superpowers/plans/2026-04-15-duckdb-migration.md

# DuckDB Migration Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Replace the in-memory global Polars dataframes in `src/utils.py` (the full `df` plus five derived frames) with an on-disk DuckDB database, reducing steady-state RSS. Per-request SQL returns small Polars frames that existing downstream code continues to consume.

**Architecture:** A new module `src/db.py` owns the DuckDB lifecycle — startup build under a `fcntl.flock` lock, atomic `os.replace` swap, read-only runtime connection with per-call cursors, and a `query_marches()` helper that returns `pl.DataFrame`. Row-level transforms stay in Polars (via `connection.register("frame", pl_frame)`) so `booleans_to_strings` and the null-name replacement remain the single source of truth. `df_acheteurs` and `df_titulaires` stay as in-memory Polars frames populated from DuckDB at import time (they feed the autocomplete search).

**Tech Stack:** Python 3.10+, DuckDB (Python API), Polars, Dash, pytest (+ dash Selenium testing), pre-commit (prettier, ruff).

**Spec:** `docs/superpowers/specs/2026-04-15-duckdb-migration-design.md`

---

## File Structure

### Created files

| Path               | Responsibility                                                                                                         |
| ------------------ | ---------------------------------------------------------------------------------------------------------------------- |
| `src/db.py`        | DuckDB lifecycle: `should_rebuild`, `build_database`, module-level `conn`, `schema`, `get_cursor()`, `query_marches()` |
| `tests/test_db.py` | Unit tests for `should_rebuild`, `build_database`, `query_marches`, and concurrent-startup lock behavior               |

### Modified files

| Path                                   | What changes                                                                                                                                                                                           |
| -------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `pyproject.toml`                       | Add `duckdb` dependency                                                                                                                                                                                |
| `.gitignore`                           | Ignore `**/decp.duckdb`, `**/decp.duckdb.tmp`, `**/decp.duckdb.lock`                                                                                                                                   |
| `src/utils.py`                         | Remove `df` global, replace `df_acheteurs` / `df_titulaires` population with DuckDB-backed queries, remove `df_*_marches` and `df_*_departement` globals, expose `schema` / `columns` from `src/db.py` |
| `src/pages/marche.py`                  | `df.lazy().filter(...)` → `query_marches("uid = ?", (uid,)).lazy()`                                                                                                                                    |
| `src/pages/acheteur.py`                | Same pattern + replace `df.collect_schema()` with module `schema`, `df.columns` with `schema.names()`                                                                                                  |
| `src/pages/titulaire.py`               | Same pattern as acheteur                                                                                                                                                                               |
| `src/pages/tableau.py`                 | `df.lazy()` → `query_marches().lazy()`, `df.columns` → `schema.names()`, `df.width` → `len(schema.names())`                                                                                            |
| `src/pages/observatoire.py`            | 3 call sites: `df.lazy()` → `query_marches().lazy()`, `df.columns` → `schema.names()`                                                                                                                  |
| `src/pages/arbre/departement.py`       | `df_acheteurs_departement` / `df_titulaires_departement` → `get_cursor().execute("SELECT ... FROM {table} WHERE ...", [...]).pl()`                                                                     |
| `src/pages/arbre/liste_marches_org.py` | `df_acheteurs_marches` / `df_titulaires_marches` → `get_cursor().execute(...).pl()`                                                                                                                    |
| `src/figures.py`                       | `df.columns` → `schema.names()` (via `src.db`)                                                                                                                                                         |
| `tests/conftest.py`                    | Ensure the DuckDB test file is rebuilt from the freshly-written `test.parquet` each session                                                                                                            |

---

## Task 1: Add DuckDB dependency and ignore patterns

**Files:**

- Modify: `pyproject.toml` (dependencies list at top)
- Modify: `.gitignore`

- [ ] **Step 1: Add `duckdb` to `pyproject.toml`**

Open `pyproject.toml`, find the `dependencies = [...]` block, and insert `"duckdb",` on a new line before `"flask-caching",`. Result:

```toml
dependencies = [
  "dash==3.4.0",
  "dash[compress]",
  "polars",
  "gunicorn",
  "dash-bootstrap-components",
  "python-dotenv",
  "xlsxwriter",
  "plotly[express]",
  "httpx",
  "pandas",
  "unidecode",
  "dash-leaflet",
  "dash-extensions",
  "duckdb",
  "flask-caching",
]
```

- [ ] **Step 2: Install the dependency**

Run: `uv pip install -e ".[dev]"`

Expected: installation completes, `python -c "import duckdb; print(duckdb.__version__)"` prints a version.

- [ ] **Step 3: Add DuckDB artifacts to `.gitignore`**

Append to `.gitignore`:

```
# DuckDB runtime artifacts (regenerated from decp_prod.parquet at startup)
**/decp.duckdb
**/decp.duckdb.tmp
**/decp.duckdb.lock
```

- [ ] **Step 4: Commit**

```bash
git add pyproject.toml .gitignore uv.lock
git commit -m "Ajout de la dépendance duckdb"
```

---

## Task 2: Write failing tests for `should_rebuild`

**Files:**

- Create: `tests/test_db.py`

- [ ] **Step 1: Write the failing tests**

Create `tests/test_db.py`:

```python
import os
import time
from pathlib import Path

import pytest


@pytest.fixture
def parquet_and_db(tmp_path, monkeypatch):
    parquet = tmp_path / "source.parquet"
    db = tmp_path / "decp.duckdb"
    parquet.write_bytes(b"fake parquet content")
    monkeypatch.delenv("REBUILD_DUCKDB", raising=False)
    monkeypatch.delenv("DEVELOPMENT", raising=False)
    return parquet, db


def test_should_rebuild_when_db_missing(parquet_and_db, monkeypatch):
    from src.db import should_rebuild
    parquet, db = parquet_and_db
    monkeypatch.setenv("DEVELOPMENT", "true")
    assert should_rebuild(db, parquet) is True


def test_should_rebuild_prod_when_parquet_newer(parquet_and_db, monkeypatch):
    from src.db import should_rebuild
    parquet, db = parquet_and_db
    db.write_bytes(b"x")
    time.sleep(0.01)
    parquet.touch()
    monkeypatch.setenv("DEVELOPMENT", "false")
    assert should_rebuild(db, parquet) is True


def test_should_not_rebuild_prod_when_parquet_older(parquet_and_db, monkeypatch):
    from src.db import should_rebuild
    parquet, db = parquet_and_db
    parquet.touch()
    time.sleep(0.01)
    db.write_bytes(b"x")
    monkeypatch.setenv("DEVELOPMENT", "false")
    assert should_rebuild(db, parquet) is False


def test_should_not_rebuild_dev_even_when_parquet_newer(parquet_and_db, monkeypatch):
    from src.db import should_rebuild
    parquet, db = parquet_and_db
    db.write_bytes(b"x")
    time.sleep(0.01)
    parquet.touch()
    monkeypatch.setenv("DEVELOPMENT", "true")
    monkeypatch.delenv("REBUILD_DUCKDB", raising=False)
    assert should_rebuild(db, parquet) is False


def test_should_rebuild_dev_when_rebuild_forced(parquet_and_db, monkeypatch):
    from src.db import should_rebuild
    parquet, db = parquet_and_db
    db.write_bytes(b"x")
    time.sleep(0.01)
    parquet.touch()
    monkeypatch.setenv("DEVELOPMENT", "true")
    monkeypatch.setenv("REBUILD_DUCKDB", "true")
    assert should_rebuild(db, parquet) is True
```

- [ ] **Step 2: Run tests and verify they fail**

Run: `uv run pytest tests/test_db.py -v`

Expected: FAIL with `ModuleNotFoundError: No module named 'src.db'` (or `ImportError`).

---

## Task 3: Implement `should_rebuild` in `src/db.py`

**Files:**

- Create: `src/db.py`

- [ ] **Step 1: Create minimal `src/db.py`**

```python
import logging
import os
from pathlib import Path

logger = logging.getLogger("decp.info")


def should_rebuild(db_path: Path, parquet_path: Path) -> bool:
    """Decide whether to rebuild the DuckDB database from the source Parquet.

    Rules:
      - Rebuild if the DuckDB file does not exist.
      - Otherwise, rebuild only if the source Parquet is newer than the DB,
        EXCEPT in development mode without REBUILD_DUCKDB=true (dev keeps
        a stable DB across reloads unless explicitly opted in).
    """
    db_path = Path(db_path)
    parquet_path = Path(parquet_path)

    if not db_path.exists():
        return True

    dev = os.getenv("DEVELOPMENT", "False").lower() == "true"
    force = os.getenv("REBUILD_DUCKDB", "False").lower() == "true"
    if dev and not force:
        return False

    return parquet_path.stat().st_mtime > db_path.stat().st_mtime
```

- [ ] **Step 2: Run tests and verify they pass**

Run: `uv run pytest tests/test_db.py -v`

Expected: 5 passed.

- [ ] **Step 3: Commit**

```bash
git add src/db.py tests/test_db.py
git commit -m "Ajout de src.db.should_rebuild et ses tests"
```

---

## Task 4: Write failing tests for `build_database` and `query_marches`

**Files:**

- Modify: `tests/test_db.py`

- [ ] **Step 1: Append build/query tests to `tests/test_db.py`**

```python
import datetime
import polars as pl


@pytest.fixture
def built_db(tmp_path, monkeypatch):
    """Build a DuckDB from a small Polars frame written as parquet."""
    parquet_path = tmp_path / "source.parquet"
    db_path = tmp_path / "decp.duckdb"

    data = pl.DataFrame(
        [
            {
                "uid": "1",
                "id": "1",
                "objet": "Travaux",
                "acheteur_id": "A1",
                "acheteur_nom": "Mairie",
                "acheteur_departement_code": "75",
                "titulaire_id": "T1",
                "titulaire_nom": "Entreprise",
                "titulaire_departement_code": "35",
                "titulaire_typeIdentifiant": "SIRET",
                "montant": 1000.0,
                "dateNotification": datetime.date(2025, 1, 1),
                "donneesActuelles": True,
                "marcheInnovant": True,
            },
            {
                "uid": "2",
                "id": "2",
                "objet": "Études",
                "acheteur_id": "A1",
                "acheteur_nom": "Mairie",
                "acheteur_departement_code": "75",
                "titulaire_id": "T2",
                "titulaire_nom": None,
                "titulaire_departement_code": "75",
                "titulaire_typeIdentifiant": "SIRET",
                "montant": 500.0,
                "dateNotification": datetime.date(2024, 6, 1),
                "donneesActuelles": True,
                "marcheInnovant": False,
            },
            {
                "uid": "3",
                "id": "3",
                "objet": "Ancien",
                "acheteur_id": "A2",
                "acheteur_nom": None,
                "acheteur_departement_code": "13",
                "titulaire_id": "T3",
                "titulaire_nom": "Autre",
                "titulaire_departement_code": "13",
                "titulaire_typeIdentifiant": "SIRET",
                "montant": 100.0,
                "dateNotification": datetime.date(2023, 1, 1),
                "donneesActuelles": False,  # must be filtered out
                "marcheInnovant": False,
            },
        ]
    )
    data.write_parquet(parquet_path)
    monkeypatch.setenv("DATA_FILE_PARQUET_PATH", str(parquet_path))

    from src.db import build_database
    build_database(db_path, parquet_path)
    return db_path


def test_build_filters_donnees_actuelles(built_db):
    import duckdb
    with duckdb.connect(str(built_db), read_only=True) as c:
        rows = c.execute("SELECT uid FROM decp ORDER BY uid").fetchall()
    assert [r[0] for r in rows] == ["1", "2"]


def test_build_converts_booleans_to_oui_non(built_db):
    import duckdb
    with duckdb.connect(str(built_db), read_only=True) as c:
        values = c.execute(
            "SELECT marcheInnovant FROM decp ORDER BY uid"
        ).fetchall()
    assert [v[0] for v in values] == ["oui", "non"]


def test_build_replaces_null_org_names(built_db):
    import duckdb
    with duckdb.connect(str(built_db), read_only=True) as c:
        titulaire_2 = c.execute(
            "SELECT titulaire_nom FROM decp WHERE uid = '2'"
        ).fetchone()
    assert titulaire_2[0] == "[Identifiant non reconnu dans la base INSEE]"


def test_build_creates_derived_tables(built_db):
    import duckdb
    with duckdb.connect(str(built_db), read_only=True) as c:
        tables = {r[0] for r in c.execute("SHOW TABLES").fetchall()}
    assert {"decp", "acheteurs_marches", "titulaires_marches",
            "acheteurs_departement", "titulaires_departement"} <= tables


def test_query_marches_returns_polars_frame(built_db, monkeypatch):
    monkeypatch.setenv("DATA_FILE_PARQUET_PATH", str(built_db.parent / "source.parquet"))
    # Force src.db to load pointing at this test DB.
    import importlib, src.db
    importlib.reload(src.db)
    from src.db import query_marches
    frame = query_marches("acheteur_id = ?", ("A1",))
    assert isinstance(frame, pl.DataFrame)
    assert frame.height == 2
    assert set(frame["uid"].to_list()) == {"1", "2"}
```

- [ ] **Step 2: Run new tests and verify they fail**

Run: `uv run pytest tests/test_db.py -v -k "build or query_marches"`

Expected: FAIL with `ImportError: cannot import name 'build_database' from 'src.db'` (or similar on `query_marches`).

---

## Task 5: Implement `build_database` in `src/db.py`

**Files:**

- Modify: `src/db.py`

- [ ] **Step 1: Add imports and `build_database`**

Replace the contents of `src/db.py` with:

```python
import fcntl
import logging
import os
from pathlib import Path

import duckdb
import polars as pl
import polars.selectors as cs
from polars.exceptions import ComputeError
from time import sleep

logger = logging.getLogger("decp.info")


def should_rebuild(db_path: Path, parquet_path: Path) -> bool:
    db_path = Path(db_path)
    parquet_path = Path(parquet_path)
    if not db_path.exists():
        return True
    dev = os.getenv("DEVELOPMENT", "False").lower() == "true"
    force = os.getenv("REBUILD_DUCKDB", "False").lower() == "true"
    if dev and not force:
        return False
    return parquet_path.stat().st_mtime > db_path.stat().st_mtime


def _load_source_frame(parquet_path: Path) -> pl.DataFrame:
    """Read the source parquet and apply the row-level transforms.

    Kept here (not in utils.py) so src.db has no dependency on utils.
    Mirrors the behavior previously in utils.get_decp_data().
    """
    try:
        lff: pl.LazyFrame = pl.scan_parquet(str(parquet_path))
    except ComputeError:
        logger.info("Lecture du parquet échouée, nouvelle tentative dans 10s...")
        sleep(10)
        lff = pl.scan_parquet(str(parquet_path))

    lff = lff.sort(by=["dateNotification", "uid"], descending=True, nulls_last=True)
    lff = lff.filter(pl.col("donneesActuelles")).drop("donneesActuelles")

    # booleans_to_strings: true → "oui", false → "non"
    lff = lff.with_columns(
        pl.col(cs.Boolean).cast(pl.String).str.replace("true", "oui").str.replace("false", "non")
    )

    for col in ["acheteur_nom", "titulaire_nom"]:
        lff = lff.with_columns(
            pl.when(pl.col(col).is_null())
            .then(pl.lit("[Identifiant non reconnu dans la base INSEE]"))
            .otherwise(pl.col(col))
            .name.keep()
        )

    return lff.collect()


def build_database(db_path: Path, parquet_path: Path) -> None:
    """Build the DuckDB database atomically under an exclusive lock.

    Caller MUST hold the fcntl.flock on the .lock file.
    """
    db_path = Path(db_path)
    parquet_path = Path(parquet_path)
    tmp_path = db_path.with_suffix(".duckdb.tmp")
    if tmp_path.exists():
        tmp_path.unlink()

    logger.info(f"Construction de la base DuckDB à partir de {parquet_path}...")
    frame = _load_source_frame(parquet_path)

    with duckdb.connect(str(tmp_path)) as w:
        w.register("frame", frame)
        w.execute("CREATE TABLE decp AS SELECT * FROM frame")
        w.execute(
            "CREATE TABLE acheteurs_marches AS "
            "SELECT DISTINCT uid, objet, acheteur_id FROM decp "
            "ORDER BY acheteur_id"
        )
        w.execute(
            "CREATE TABLE titulaires_marches AS "
            "SELECT DISTINCT uid, objet, titulaire_id FROM decp "
            "ORDER BY titulaire_id"
        )
        w.execute(
            "CREATE TABLE acheteurs_departement AS "
            "SELECT DISTINCT acheteur_id, acheteur_nom, acheteur_departement_code "
            "FROM decp ORDER BY acheteur_nom"
        )
        w.execute(
            "CREATE TABLE titulaires_departement AS "
            "SELECT DISTINCT titulaire_id, titulaire_nom, titulaire_departement_code "
            "FROM decp ORDER BY titulaire_nom"
        )

    os.replace(tmp_path, db_path)
    logger.info(f"Base DuckDB construite : {db_path}")
```

- [ ] **Step 2: Run build tests and verify they pass**

Run: `uv run pytest tests/test_db.py -v -k "build"`

Expected: 4 passed (`test_build_filters_donnees_actuelles`, `test_build_converts_booleans_to_oui_non`, `test_build_replaces_null_org_names`, `test_build_creates_derived_tables`).

---

## Task 6: Implement startup guard, read-only connection, `schema`, `get_cursor`, `query_marches`

**Files:**

- Modify: `src/db.py`

- [ ] **Step 1: Append startup logic and query helpers**

Add at the bottom of `src/db.py`:

```python
def _resolve_db_path() -> Path:
    parquet = os.getenv("DATA_FILE_PARQUET_PATH")
    if not parquet:
        raise RuntimeError("DATA_FILE_PARQUET_PATH is not set")
    return Path(parquet).parent / "decp.duckdb"


def _ensure_database() -> Path:
    db_path = _resolve_db_path()
    parquet_path = Path(os.getenv("DATA_FILE_PARQUET_PATH"))
    lock_path = db_path.with_suffix(".duckdb.lock")

    with open(lock_path, "w") as lock_fd:
        fcntl.flock(lock_fd, fcntl.LOCK_EX)
        if should_rebuild(db_path, parquet_path):
            build_database(db_path, parquet_path)
    return db_path


DB_PATH = _ensure_database()
conn: duckdb.DuckDBPyConnection = duckdb.connect(str(DB_PATH), read_only=True)
schema: pl.Schema = conn.execute("SELECT * FROM decp LIMIT 0").pl().schema


def get_cursor() -> duckdb.DuckDBPyConnection:
    """Return a per-request cursor that shares the process-wide connection."""
    return conn.cursor()


def query_marches(
    where_sql: str = "TRUE",
    params: tuple = (),
    columns: list[str] | None = None,
    order_by: str | None = None,
    limit: int | None = None,
) -> pl.DataFrame:
    """Run a parameterized SELECT against the decp table and return Polars.

    `where_sql` and `order_by` are trusted SQL fragments (callers are internal
    code, never user input). `params` values are passed through DuckDB's
    parameter binding.
    """
    cols = ", ".join(columns) if columns else "*"
    sql = f"SELECT {cols} FROM decp WHERE {where_sql}"
    if order_by:
        sql += f" ORDER BY {order_by}"
    if limit is not None:
        sql += f" LIMIT {int(limit)}"
    return get_cursor().execute(sql, list(params)).pl()
```

- [ ] **Step 2: Run the full `test_db.py` suite**

Run: `uv run pytest tests/test_db.py -v`

Expected: all tests pass (9 total).

- [ ] **Step 3: Commit**

```bash
git add src/db.py tests/test_db.py
git commit -m "Implémentation de build_database, query_marches et connexion DuckDB"
```

---

## Task 7: Write failing test for concurrent build locking

**Files:**

- Modify: `tests/test_db.py`

- [ ] **Step 1: Append a lock-serialization test**

```python
import threading


def test_concurrent_build_serialized(tmp_path, monkeypatch):
    """Two workers starting at once must not stomp on each other's tmp file."""
    parquet_path = tmp_path / "source.parquet"
    db_path = tmp_path / "decp.duckdb"

    pl.DataFrame(
        [
            {
                "uid": "1", "id": "1", "objet": "x",
                "acheteur_id": "A", "acheteur_nom": "n",
                "titulaire_id": "T", "titulaire_nom": "n",
                "acheteur_departement_code": "75",
                "titulaire_departement_code": "75",
                "titulaire_typeIdentifiant": "SIRET",
                "montant": 1.0,
                "dateNotification": datetime.date(2025, 1, 1),
                "donneesActuelles": True,
                "marcheInnovant": False,
            }
        ]
    ).write_parquet(parquet_path)

    monkeypatch.setenv("DATA_FILE_PARQUET_PATH", str(parquet_path))

    from src.db import build_database, should_rebuild
    import fcntl

    errors: list[Exception] = []

    def worker():
        try:
            lock_path = db_path.with_suffix(".duckdb.lock")
            with open(lock_path, "w") as lock_fd:
                fcntl.flock(lock_fd, fcntl.LOCK_EX)
                if should_rebuild(db_path, parquet_path):
                    build_database(db_path, parquet_path)
        except Exception as exc:
            errors.append(exc)

    threads = [threading.Thread(target=worker) for _ in range(3)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    assert errors == []
    assert db_path.exists()
    assert not db_path.with_suffix(".duckdb.tmp").exists()
```

- [ ] **Step 2: Run it**

Run: `uv run pytest tests/test_db.py::test_concurrent_build_serialized -v`

Expected: PASS (the lock and `should_rebuild` short-circuit inside the worker already implements correct serialization — this test verifies the contract, no code changes needed).

If it fails, fix the issue in `src/db.py` (most likely: `build_database` does not clean up `tmp_path` when re-entered). Then re-run.

- [ ] **Step 3: Commit**

```bash
git add tests/test_db.py
git commit -m "Test de sérialisation de la construction concurrente de la base"
```

---

## Task 8: Make `tests/conftest.py` rebuild the DuckDB with the test parquet

**Files:**

- Modify: `tests/conftest.py`
- Modify: `pyproject.toml` (pytest env block)

- [ ] **Step 1: Force a fresh DuckDB rebuild in the test session**

The existing `test_data` fixture writes `tests/test.parquet` at session start. We must delete any leftover `tests/decp.duckdb` before `src.db` is imported by the app, because `src.db` is imported at module load by `src/utils.py`, which is in turn imported by pages.

Update `tests/conftest.py`:

```python
import datetime
import os
from pathlib import Path

import polars as pl
import pytest
from selenium.webdriver.chrome.options import Options


@pytest.fixture(scope="session", autouse=True)
def test_data():
    data = [
        {
            "uid": "1",
            "id": "1",
            "acheteur_nom": "ACHETEUR 1",
            "acheteur_id": "123",
            "titulaire_nom": "TITULAIRE 1",
            "titulaire_id": "345",
            "montant": 10,
            "dateNotification": datetime.date(2025, 1, 1),
            "codeCPV": "71600000",
            "donneesActuelles": True,
            "acheteur_departement_code": "75",
            "acheteur_departement_nom": "Paris",
            "acheteur_commune_nom": "Paris",
            "titulaire_departement_code": "35",
            "titulaire_departement_nom": "Ille-et-Vilaine",
            "titulaire_commune_nom": "Rennes",
            "titulaire_distance": 10,
            "titulaire_typeIdentifiant": "SIRET",
            "objet": "Objet test",
            "dureeRestanteMois": 12,
            "lieuExecution_code": "75001",
            "sourceFile": "test.xml",
            "sourceDataset": "test_dataset",
            "datePublicationDonnees": datetime.date(2025, 1, 1),
            "considerationsSociales": "",
            "considerationsEnvironnementales": "",
            "type": "Marché",
            "acheteur_categorie": "Collectivité",
            "titulaire_categorie": "PME",
        }
    ]
    parquet_path = Path(os.path.abspath("tests/test.parquet"))
    db_path = parquet_path.parent / "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
    # the freshly-written parquet at import time.
    for artifact in (db_path, db_path.with_suffix(".duckdb.tmp")):
        if artifact.exists():
            artifact.unlink()

    yield str(parquet_path)


def pytest_setup_options():
    options = Options()
    options.add_argument("--window-size=1200,1200 ")
    options.add_experimental_option(
        "prefs",
        {
            "download.default_directory": "/home/colin/git/decp.info",
            "download.prompt_for_download": False,
            "download.directory_upgrade": True,
            "safebrowsing.enabled": True,
        },
    )
    return options
```

- [ ] **Step 2: Verify existing tests still import cleanly**

Run: `uv run pytest tests/test_db.py -v`

Expected: all 10 DB tests still pass (the conftest changes only affect integration tests).

- [ ] **Step 3: Commit**

```bash
git add tests/conftest.py
git commit -m "Suppression de la base DuckDB de test obsolète avant chaque session"
```

---

## Task 9: Migrate `src/utils.py` — introduce `src.db` imports, keep old globals

**Files:**

- Modify: `src/utils.py`

This task does **not** remove the old globals yet. It wires `src.db` in and re-points `schema` / `columns` so pages can start using them without a flag day.

- [ ] **Step 1: Import from `src.db` at the top of `utils.py`**

After the `from unidecode import unidecode` line, add:

```python
from src.db import conn as duckdb_conn  # noqa: F401  (exposed for convenience)
from src.db import get_cursor, query_marches, schema  # noqa: F401
```

- [ ] **Step 2: Replace the module-level assignments at the bottom**

Find lines 891–913 (the block starting with `df: pl.DataFrame = get_decp_data()`).

Replace the block with:

```python
df: pl.DataFrame = get_decp_data()
# schema and columns now come from src.db; overwrite in case any local code
# still reads them directly from utils.
schema = schema  # re-exported from src.db
columns = schema.names()

df_acheteurs = get_org_data(df, "acheteur")
df_titulaires = get_org_data(df, "titulaire")
df_acheteurs_departement: pl.DataFrame = (
    df_acheteurs.select(["acheteur_id", "acheteur_nom", "acheteur_departement_code"])
    .unique()
    .sort("acheteur_nom")
)
df_titulaires_departement: pl.DataFrame = (
    df_titulaires.select(
        ["titulaire_id", "titulaire_nom", "titulaire_departement_code"]
    )
    .unique()
    .sort("titulaire_nom")
)
df_acheteurs_marches: pl.DataFrame = (
    df.select("uid", "objet", "acheteur_id").unique().sort("acheteur_id")
)
df_titulaires_marches: pl.DataFrame = (
    df.select("uid", "objet", "titulaire_id").unique().sort("titulaire_id")
)
```

No removals — we keep the old globals alive. Pages that already use `schema` / `columns` now transparently get them from `src.db`.

- [ ] **Step 3: Run the Selenium test suite**

Run: `uv run pytest tests/test_main.py -v`

Expected: all tests pass. The app loads; nothing has changed behaviorally yet.

- [ ] **Step 4: Commit**

```bash
git add src/utils.py
git commit -m "Intégration de src.db dans utils (coexistence avec les globaux)"
```

---

## Task 10: Migrate `src/pages/marche.py` (1 call site)

**Files:**

- Modify: `src/pages/marche.py`

- [ ] **Step 1: Replace `df` with `query_marches`**

Edit the import block:

```python
from src.utils import (
    data_schema,
    format_values,
    make_org_jsonld,
    meta_content,
    unformat_montant,
)
from src.db import query_marches
```

Replace the body of `get_marche_data`:

```python
def get_marche_data(url) -> tuple[dict, list]:
    marche_uid = url.split("/")[-1]

    # Filtre SQL côté DuckDB, puis Polars pour le post-traitement
    dff_marche = query_marches("uid = ?", (marche_uid,))
    if dff_marche.height == 0:
        return {}, []

    lff = dff_marche.lazy()
    dff_titulaires = lff.select(cs.starts_with("titulaire")).collect(engine="streaming")
    dff_marche_unique = lff.unique("uid").collect(engine="streaming")
    dff_marche_unique = format_values(dff_marche_unique)

    return dff_marche_unique.to_dicts()[0], dff_titulaires.to_dicts()
```

- [ ] **Step 2: Smoke-test**

Run: `uv run pytest tests/test_main.py -v`

Expected: all tests pass.

Manually: `uv run run.py`, navigate to `/marches/<any_uid_from_test_data>` (e.g. `/marches/1`), confirm page renders with the buyer, amount, and titulaire list.

- [ ] **Step 3: Commit**

```bash
git add src/pages/marche.py
git commit -m "marche.py : utilisation de query_marches au lieu du df global"
```

---

## Task 11: Migrate `src/pages/acheteur.py`

**Files:**

- Modify: `src/pages/acheteur.py`

- [ ] **Step 1: Replace `df` imports**

Update the import block:

```python
from src.utils import (
    columns,
    df_acheteurs,
    filter_table_data,
    format_number,
    get_annuaire_data,
    get_button_properties,
    get_default_hidden_columns,
    # ... any other existing non-df imports
)
from src.db import query_marches, schema
```

(Remove `df` from the `src.utils` import list.)

- [ ] **Step 2: Replace `df.columns` at line 73**

```python
columns=[{"id": col, "name": col} for col in schema.names()],
```

- [ ] **Step 3: Replace `df.collect_schema()` at line 303**

```python
dff = pl.DataFrame(schema=schema)
```

- [ ] **Step 4: Replace `df.lazy().filter(...)` in `get_acheteur_marches_data`**

```python
def get_acheteur_marches_data(url, ach_year: str) -> tuple:
    acheteur_siret = url.split("/")[-1]
    lff = query_marches("acheteur_id = ?", (acheteur_siret,)).lazy()
    if ach_year and ach_year != "Toutes les années":
        ach_year = int(ach_year)
        lff = lff.filter(pl.col("dateNotification").dt.year() == ach_year)
    lff = lff.sort(["dateNotification", "uid"], descending=True, nulls_last=True)
    dff: pl.DataFrame = lff.collect(engine="streaming")
    download_disabled, download_text, download_title = get_button_properties(dff.height)
    data = dff.to_dicts()
    return data, download_disabled, download_text, download_title
```

- [ ] **Step 5: Run tests and smoke-test**

Run: `uv run pytest tests/test_main.py -v`

Manually: `uv run run.py`, navigate to `/acheteurs/123` (test data SIRET), confirm KPIs, charts, and table render.

- [ ] **Step 6: Commit**

```bash
git add src/pages/acheteur.py
git commit -m "acheteur.py : migration vers query_marches et schema depuis src.db"
```

---

## Task 12: Migrate `src/pages/titulaire.py`

**Files:**

- Modify: `src/pages/titulaire.py`

- [ ] **Step 1: Replace imports and call sites**

Remove `df` from the `src.utils` import list. Add:

```python
from src.db import query_marches, schema
```

- [ ] **Step 2: Replace `df.columns` at line 72**

```python
columns=[{"id": col, "name": col} for col in schema.names()],
```

- [ ] **Step 3: Replace `df.lazy().filter(...)` in `get_titulaire_marches_data`**

```python
def get_titulaire_marches_data(url, titulaire_year: str) -> tuple:
    titulaire_siret = url.split("/")[-1]
    lff = query_marches(
        "titulaire_id = ? AND titulaire_typeIdentifiant = 'SIRET'",
        (titulaire_siret,),
    ).lazy()
    if titulaire_year and titulaire_year != "Toutes les années":
        lff = lff.filter(
            pl.col("dateNotification").cast(pl.String).str.starts_with(titulaire_year)
        )
    lff = lff.sort(["dateNotification", "uid"], descending=True, nulls_last=True)
    lff = lff.fill_null("")
    dff: pl.DataFrame = lff.collect(engine="streaming")
    # ... rest of function unchanged
```

- [ ] **Step 4: Run tests and smoke-test**

Run: `uv run pytest tests/test_main.py -v`

Manually: `uv run run.py`, navigate to `/titulaires/345` (test SIRET), confirm it renders.

- [ ] **Step 5: Commit**

```bash
git add src/pages/titulaire.py
git commit -m "titulaire.py : migration vers query_marches et schema depuis src.db"
```

---

## Task 13: Migrate `src/pages/arbre/departement.py`

**Files:**

- Modify: `src/pages/arbre/departement.py`

- [ ] **Step 1: Replace imports and queries**

```python
import polars as pl
from dash import Input, Output, callback, dcc, html, register_page

from src.utils import departements
from src.db import get_cursor

# ... (register_page and layout unchanged) ...


@callback(
    Output(component_id="departement_marches", component_property="children"),
    Input(component_id="departement_url", component_property="pathname"),
)
def departement_marches(url):
    departement = url.split("/")[-1]

    def make_link_list(org_type) -> list:
        table = (
            "acheteurs_departement" if org_type == "acheteur"
            else "titulaires_departement" if org_type == "titulaire"
            else None
        )
        if table is None:
            raise ValueError
        col_prefix = org_type
        rows = get_cursor().execute(
            f"SELECT {col_prefix}_id, {col_prefix}_nom "
            f"FROM {table} "
            f"WHERE {col_prefix}_departement_code = ? "
            f"ORDER BY {col_prefix}_nom",
            [departement],
        ).fetchall()

        link_list = []
        for org_id, org_nom in rows:
            li = html.Li(
                [
                    dcc.Link(
                        org_nom,
                        href=url + f"/{org_type}/{org_id}",
                        title=f"Marchés publics de {org_nom}",
                    ),
                    " ",
                    dcc.Link(
                        "(page dédiée)",
                        href=f"/{org_type}s/{org_id}",
                        title=f"Page dédiée aux marchés publics de {org_nom}",
                    ),
                ]
            )
            link_list.append(li)
        return link_list

    content = [
        html.H3("Acheteurs publics du département"),
        html.Ul(make_link_list("acheteur")),
        html.H3("Titulaires du département"),
        html.Ul(make_link_list("titulaire")),
    ]
    return content
```

- [ ] **Step 2: Run tests and smoke-test**

Run: `uv run pytest tests/test_main.py -v`

Manually: `uv run run.py`, navigate to `/departements/75`, confirm list of acheteurs and titulaires renders.

- [ ] **Step 3: Commit**

```bash
git add src/pages/arbre/departement.py
git commit -m "arbre/departement.py : requête DuckDB directe pour les listes départementales"
```

---

## Task 14: Migrate `src/pages/arbre/liste_marches_org.py`

**Files:**

- Modify: `src/pages/arbre/liste_marches_org.py`

- [ ] **Step 1: Replace imports and queries**

```python
import polars as pl
from dash import Input, Output, callback, dcc, html, register_page

from src.utils import df_acheteurs, df_titulaires
from src.db import get_cursor

name = "Liste des marchés publics"


def make_org_nom_verbe(org_type, org_id) -> tuple:
    if org_type == "titulaire":
        source = df_titulaires
        verbe = "remportés"
    elif org_type == "acheteur":
        source = df_acheteurs
        verbe = "attribués"
    else:
        raise ValueError

    org_nom = (
        source.filter(pl.col(f"{org_type}_id") == org_id)
        .select(f"{org_type}_nom")
        .item(0, 0)
    )
    return org_nom, verbe


# ... (register_page / layout unchanged) ...


@callback(
    Output(component_id="liste_marches", component_property="children"),
    Input(component_id="liste_marches_url", component_property="pathname"),
)
def liste_marches(url):
    org_type = url.split("/")[-2]
    org_id = url.split("/")[-1]

    def make_link_list() -> list:
        table = (
            "acheteurs_marches" if org_type == "acheteur"
            else "titulaires_marches" if org_type == "titulaire"
            else None
        )
        if table is None:
            raise ValueError
        rows = get_cursor().execute(
            f"SELECT uid, objet FROM {table} WHERE {org_type}_id = ?",
            [org_id],
        ).fetchall()

        return [
            html.Li(
                dcc.Link(
                    objet,
                    href=f"/marches/{uid}",
                    title=f"Marchés public attribué : {objet}",
                )
            )
            for uid, objet in rows
        ]

    nom, verbe = make_org_nom_verbe(org_type, org_id)
    return [
        html.H3(f"Marchés publics {verbe} par {nom}"),
        html.Ul(make_link_list()),
    ]
```

- [ ] **Step 2: Run tests and smoke-test**

Run: `uv run pytest tests/test_main.py -v`

Manually: `uv run run.py`, navigate to `/departements/75/acheteur/123`, confirm list of marchés renders.

- [ ] **Step 3: Commit**

```bash
git add src/pages/arbre/liste_marches_org.py
git commit -m "arbre/liste_marches_org.py : requêtes DuckDB pour les listes de marchés"
```

---

## Task 15: Migrate `src/pages/tableau.py`

**Files:**

- Modify: `src/pages/tableau.py`

- [ ] **Step 1: Replace imports**

Remove `df` from `src.utils` imports; remove `schema` from `src.utils` imports if present, re-import it from `src.db`:

```python
from src.utils import (
    columns,
    filter_table_data,
    get_default_hidden_columns,
    invert_columns,
    logger,
    meta_content,
    prepare_table_data,
    sort_table_data,
    update_date_iso,
)
from src.db import query_marches, schema
```

- [ ] **Step 2: Replace `df.columns` at line 64**

```python
columns=[{"id": col, "name": col} for col in schema.names()],
```

- [ ] **Step 3: Replace `df.width` at line 131**

In the `dcc.Markdown` block, replace `{str(df.width)}` with `{len(schema.names())}`.

- [ ] **Step 4: Replace `df.lazy()` at line 319 in `download_data`**

```python
def download_data(n_clicks, filter_query, sort_by, hidden_columns: list = None):
    lff: pl.LazyFrame = query_marches().lazy()

    if hidden_columns:
        lff = lff.drop(hidden_columns)

    if filter_query:
        lff = filter_table_data(lff, filter_query, "tab download")

    if sort_by and len(sort_by) > 0:
        lff = sort_table_data(lff, sort_by)

    def to_bytes(buffer):
        lff.collect(engine="streaming").write_excel(buffer, worksheet="DECP")

    date = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
    return dcc.send_bytes(to_bytes, filename=f"decp_{date}.xlsx")
```

Note: `query_marches()` with no arguments returns the full table into a Polars frame, which is then lazily filtered. This is equivalent to today's `df.lazy()` — memory profile is identical to the current path during download. We accept this: the download endpoint is infrequent and downloads are bounded by `filter_query` in practice. A future optimization could push the filter into SQL, but it requires reimplementing `filter_table_data` in SQL. Out of scope.

- [ ] **Step 5: Run tests and smoke-test**

Run: `uv run pytest tests/test_main.py -v`

Manually: `uv run run.py`, navigate to `/tableau`, apply a filter, try the download button.

- [ ] **Step 6: Commit**

```bash
git add src/pages/tableau.py
git commit -m "tableau.py : migration vers query_marches et schema depuis src.db"
```

---

## Task 16: Migrate `src/pages/observatoire.py`

**Files:**

- Modify: `src/pages/observatoire.py`

- [ ] **Step 1: Replace imports**

Remove `df` from `src.utils` imports; add `from src.db import query_marches, schema`. The `columns` import from `src.utils` stays (it already resolves to `schema.names()` via Task 9).

- [ ] **Step 2: Replace `df.columns` at line 75**

Wherever `df.columns` is used, replace with `schema.names()`.

- [ ] **Step 3: Replace `df.lazy()` at lines 668, 793, 884**

Each call site currently reads `df.lazy()`. Replace with `query_marches().lazy()`:

```python
# Line ~668
lff: pl.LazyFrame = query_marches().lazy()
lff = prepare_dashboard_data(lff=lff, **filter_params)

# Line ~793
lff = prepare_dashboard_data(lff=query_marches().lazy(), **(filter_params or {}))

# Line ~884
lff = prepare_dashboard_data(lff=query_marches().lazy(), **(filter_params or {}))
```

Same caveat as `tableau.py` download: this materializes the full table before filtering in Polars. Since `observatoire` results are already cached via `@cache.memoize(timeout=3600)`, the rebuild cost is amortized. Future optimization (pushing `prepare_dashboard_data` filters into SQL) is tracked separately and out of scope.

- [ ] **Step 4: Run tests and smoke-test**

Run: `uv run pytest tests/test_main.py -v`

Manually: `uv run run.py`, navigate to `/observatoire`, apply and remove filters, confirm charts render.

- [ ] **Step 5: Commit**

```bash
git add src/pages/observatoire.py
git commit -m "observatoire.py : migration vers query_marches et schema depuis src.db"
```

---

## Task 17: Migrate `src/figures.py`

**Files:**

- Modify: `src/figures.py`

- [ ] **Step 1: Remove `df` from the `src.utils` import block**

```python
from src.utils import (
    add_links,
    data_schema,
    departements_geojson,
    format_number,
    setup_table_columns,
)
from src.db import schema
```

- [ ] **Step 2: Replace `df.columns` at line 777**

```python
for col in schema.names()
```

- [ ] **Step 3: Run tests and smoke-test**

Run: `uv run pytest tests/test_main.py -v`

Manually: `uv run run.py`, exercise pages that render charts (`/acheteurs/123`, `/observatoire`).

- [ ] **Step 4: Commit**

```bash
git add src/figures.py
git commit -m "figures.py : suppression de l'import df global, usage de schema depuis src.db"
```

---

## Task 18: Remove the in-memory globals from `src/utils.py`

**Files:**

- Modify: `src/utils.py`

No page now references `df`, `df_acheteurs_departement`, `df_titulaires_departement`, `df_acheteurs_marches`, `df_titulaires_marches` (verify with grep below). `df_acheteurs` and `df_titulaires` remain (homepage search).

- [ ] **Step 1: Verify no consumers remain**

Run:

```bash
rg '\bdf\b|df_acheteurs_departement|df_titulaires_departement|df_acheteurs_marches|df_titulaires_marches' src/
```

Expected: only matches inside `src/utils.py` (the globals themselves and `get_decp_data`/`get_org_data`) and no matches in `src/pages/` or `src/figures.py`.

- [ ] **Step 2: Repoint `df_acheteurs` / `df_titulaires` to DuckDB**

In `src/utils.py`, replace the bottom-of-file block (lines 891–913 in the current file) with:

```python
# df_acheteurs / df_titulaires sont conservés en mémoire pour alimenter
# la recherche sur la page d'accueil (autocomplétion, filtrage par sous-chaîne
# à chaque frappe). Les colonnes reproduisent la sortie historique de
# get_org_data(df, org_type).
def _build_org_frame(org_type: str) -> pl.DataFrame:
    org_cols = [
        c for c in schema.names()
        if c.startswith(f"{org_type}_")
        and c not in (f"{org_type}_latitude", f"{org_type}_longitude")
    ]
    select_list = ", ".join(org_cols)
    group_list = ", ".join(org_cols)
    sql = (
        f"SELECT {select_list}, COUNT(*) AS \"Marchés\" "
        f"FROM decp GROUP BY {group_list}"
    )
    return get_cursor().execute(sql).pl()


df_acheteurs = _build_org_frame("acheteur")
df_titulaires = _build_org_frame("titulaire")

columns = schema.names()
```

- [ ] **Step 3: Delete now-unused functions**

Remove `get_decp_data` (lines 234–272) — it has been inlined into `src.db._load_source_frame`. Remove `get_org_data` (lines 275–284) — replaced by `_build_org_frame`.

- [ ] **Step 4: Remove the `from polars.exceptions import ComputeError` import if no longer used**

Run:

```bash
rg 'ComputeError' src/utils.py
```

If empty, remove the import line near the top of `utils.py`.

Same check for `from time import localtime, sleep`:

```bash
rg '\bsleep\b|\blocaltime\b' src/utils.py
```

Remove anything unused.

- [ ] **Step 5: Run the full test suite**

Run: `uv run pytest -v`

Expected: all tests pass.

- [ ] **Step 6: Manual smoke test of every page**

```bash
uv run run.py
```

Visit in a browser:

- `/` (recherche) — type in the acheteur and titulaire search fields, confirm autocomplete works.
- `/acheteurs/123` — confirm renders.
- `/titulaires/345` — confirm renders.
- `/marches/1` — confirm renders.
- `/tableau` — filter, sort, download.
- `/observatoire` — apply filters, confirm charts.
- `/departements` and `/departements/75` — confirm lists.
- `/departements/75/acheteur/123` — confirm marchés list.

- [ ] **Step 7: Commit**

```bash
git add src/utils.py
git commit -m "Suppression des dataframes globaux remplacés par DuckDB"
```

---

## Task 19: Measure memory impact

**Files:** none (documentation commit only)

- [ ] **Step 1: Measure RSS before (from a reference point on `main`)**

Check out `main` in a separate worktree, start the app pointing at `decp_prod.parquet`, wait for import to finish, then:

```bash
ps -o rss= -p $(pgrep -f "gunicorn app:server" | head -1)
```

Record the value (expect ~several GB with 1.5M rows in memory).

- [ ] **Step 2: Measure RSS after (on this branch)**

Back on the feature branch, same command. Record the value.

- [ ] **Step 3: Document the result**

Append a `## Outcome` section to the spec file `docs/superpowers/specs/2026-04-15-duckdb-migration-design.md` with the two RSS measurements and the ratio.

- [ ] **Step 4: Commit**

```bash
git add docs/superpowers/specs/2026-04-15-duckdb-migration-design.md
git commit -m "Mesure de l'impact mémoire post-migration DuckDB"
```

---

## Task 20: Final verification and branch readiness

- [ ] **Step 1: Run full test suite clean**

```bash
uv run pytest -v
```

Expected: all green.

- [ ] **Step 2: Run pre-commit hooks on all files**

```bash
pre-commit run --all-files
```

Expected: all hooks pass.

- [ ] **Step 3: Confirm no stray references remain**

```bash
rg '\bdf\b' src/ | rg -v 'dff|df_acheteurs|df_titulaires|data_schema|self\.df'
```

Expected: no results (or only matches inside comments / docstrings).

- [ ] **Step 4: Verify the DB file is ignored by git**

```bash
git status --ignored | rg duckdb
```

Expected: `decp.duckdb` (and `.tmp`, `.lock` if present) listed under ignored files.

Ready for PR / merge via the `finishing-a-development-branch` skill.

---

## Self-Review Notes

- **Spec coverage:** every section of `docs/superpowers/specs/2026-04-15-duckdb-migration-design.md` maps to a task:

  - Goal / Approach summary → Task 6 (`query_marches`), Tasks 10–17 (migration), Task 18 (in-memory helpers).
  - Cache invalidation rule → Tasks 2–3 (`should_rebuild` + tests).
  - Concurrency → Tasks 5–7 (`build_database` + lock test).
  - Build logic → Task 5 (reuses `_load_source_frame` with Polars transforms).
  - Module layout — `src/db.py` public surface → Task 6.
  - Configuration (`DATA_FILE_PARQUET_PATH` reused; no new env var except `REBUILD_DUCKDB`) → Task 6 (`_resolve_db_path`).
  - Testing → Tasks 2, 4, 7, 8.
  - Migration order → Tasks 10 (marche), 11 (acheteur), 12 (titulaire), 13–14 (arbre), 15 (tableau), 16 (observatoire), 17 (figures), 18 (remove globals).
  - Out-of-scope items (cache.py, observatoire-localstorage work, parquet schema, SQL views) → not touched.

- **Type consistency:** `schema` is `pl.Schema` everywhere; `conn` is `duckdb.DuckDBPyConnection`; `query_marches` signature matches the spec. `_load_source_frame` is internal to `src.db` (single underscore, not exported).

- **Placeholder scan:** no "TBD", no "similar to task N"; each step shows complete code or exact commands.