From 077af6dd2eb0968593ada549b9ce4719e7fb1832 Mon Sep 17 00:00:00 2001 From: Colin Maudry Date: Wed, 15 Apr 2026 16:16:12 +0200 Subject: [PATCH] =?UTF-8?q?Impl=C3=A9mentation=20de=20build=5Fdatabase=20a?= =?UTF-8?q?vec=20transforms=20Polars=20et=20tables=20d=C3=A9riv=C3=A9es?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _load_source_frame reprend la pipeline Polars (sort, filtre donneesActuelles, booleans_to_strings, remplacement des noms null) - build_database utilise write_parquet + read_parquet pour zéro-dépendance pyarrow (pyarrow absent du venv) et écrit atomiquement via .tmp + os.replace - 4 tables dérivées : acheteurs_marches, titulaires_marches, acheteurs_departement, titulaires_departement refs #71 Co-Authored-By: Claude Opus 4.6 --- src/db.py | 103 ++++++++++++++++++++++++++++++++++---- tests/test_db.py | 128 +++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 220 insertions(+), 11 deletions(-) diff --git a/src/db.py b/src/db.py index 4dea532..d2e2cb0 100644 --- a/src/db.py +++ b/src/db.py @@ -1,28 +1,109 @@ import logging import os from pathlib import Path +from time import sleep + +import duckdb +import polars as pl +import polars.selectors as cs +from polars.exceptions import ComputeError 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 + + +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") + staging_parquet = db_path.with_suffix(".staging.parquet") + 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) + + # Write transformed frame as parquet so DuckDB can read it natively + # (avoids pyarrow dependency for the Polars→DuckDB handoff) + frame.write_parquet(str(staging_parquet)) + try: + with duckdb.connect(str(tmp_path)) as w: + w.execute( + f"CREATE TABLE decp AS SELECT * FROM read_parquet('{staging_parquet}')" + ) + 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" + ) + finally: + if staging_parquet.exists(): + staging_parquet.unlink() + + os.replace(tmp_path, db_path) + logger.info(f"Base DuckDB construite : {db_path}") diff --git a/tests/test_db.py b/tests/test_db.py index f03d03a..2329273 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -1,6 +1,8 @@ +import datetime import os import time +import polars as pl import pytest from src.db import should_rebuild @@ -65,3 +67,129 @@ def test_should_rebuild_dev_when_rebuild_forced(parquet_and_db, monkeypatch): monkeypatch.setenv("DEVELOPMENT", "true") monkeypatch.setenv("REBUILD_DUCKDB", "true") assert should_rebuild(db, parquet) is True + + +@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 + + +@pytest.mark.skip(reason="implemented in Task 6") +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 + + import 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"}