API: parser de filtres et génération SQL paramétré (#78)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/api/filters.py

@@ -0,0 +1,138 @@
+from datetime import date, datetime
+
+import polars as pl
+
+OPERATORS = {
+    "exact",
+    "contains",
+    "notcontains",
+    "less",
+    "greater",
+    "strictly_less",
+    "strictly_greater",
+    "in",
+    "notin",
+    "isnull",
+    "isnotnull",
+    "sort",
+}
+
+RESERVED_PARAMS = {"page", "page_size", "columns", "count"}
+
+
+class FilterError(ValueError):
+    def __init__(self, message: str, field: str | None = None):
+        super().__init__(message)
+        self.field = field
+
+
+def _coerce(value: str, dtype: pl.DataType, key: str):
+    if dtype == pl.String:
+        return value
+    if dtype.is_integer():
+        try:
+            return int(value)
+        except ValueError:
+            raise FilterError(f"Valeur entière attendue, reçu {value!r}", field=key)
+    if dtype.is_float():
+        try:
+            return float(value)
+        except ValueError:
+            raise FilterError(f"Valeur décimale attendue, reçu {value!r}", field=key)
+    if dtype == pl.Date:
+        try:
+            return date.fromisoformat(value)
+        except ValueError:
+            raise FilterError(
+                f"Date ISO 8601 attendue (YYYY-MM-DD), reçu {value!r}", field=key
+            )
+    if dtype == pl.Datetime:
+        try:
+            return datetime.fromisoformat(value)
+        except ValueError:
+            raise FilterError(f"Datetime ISO 8601 attendu, reçu {value!r}", field=key)
+    return value
+
+
+def _split_key(key: str) -> tuple[str, str] | None:
+    if "__" not in key:
+        return None
+    col, _, op = key.rpartition("__")
+    if not col or not op:
+        return None
+    return col, op
+
+
+def build_where(
+    args: list[tuple[str, str]], schema: pl.Schema
+) -> tuple[str, list, str | None]:
+    """Parse query params into (where_sql, params, order_by_sql).
+
+    args: list of (key, value) tuples preserving URL order (Werkzeug MultiDict
+    preserves insertion order on `request.args.items(multi=True)`).
+    """
+    where_parts: list[str] = []
+    params: list = []
+    order_parts: list[str] = []
+
+    for key, value in args:
+        if key in RESERVED_PARAMS:
+            continue
+
+        parsed = _split_key(key)
+        if not parsed:
+            raise FilterError(f"Paramètre non reconnu : {key}", field=key)
+        col, op = parsed
+
+        if op not in OPERATORS:
+            raise FilterError(f"Opérateur inconnu : __{op}", field=key)
+
+        if col not in schema:
+            raise FilterError(f"Colonne inconnue : {col!r}", field=key)
+
+        if op == "sort":
+            direction = value.lower()
+            if direction not in ("asc", "desc"):
+                raise FilterError(
+                    f"Tri attendu 'asc' ou 'desc', reçu {value!r}", field=key
+                )
+            order_parts.append(f'"{col}" {direction.upper()}')
+            continue
+
+        if op in ("isnull", "isnotnull"):
+            sql = "IS NULL" if op == "isnull" else "IS NOT NULL"
+            where_parts.append(f'"{col}" {sql}')
+            continue
+
+        dtype = schema[col]
+
+        if op in ("in", "notin"):
+            values = [_coerce(v.strip(), dtype, key) for v in value.split(",")]
+            placeholders = ",".join(["?"] * len(values))
+            sql_op = "IN" if op == "in" else "NOT IN"
+            where_parts.append(f'"{col}" {sql_op} ({placeholders})')
+            params.extend(values)
+            continue
+
+        v = _coerce(value, dtype, key)
+
+        op_sql = {
+            "exact": "=",
+            "less": "<=",
+            "greater": ">=",
+            "strictly_less": "<",
+            "strictly_greater": ">",
+        }
+        if op in op_sql:
+            where_parts.append(f'"{col}" {op_sql[op]} ?')
+            params.append(v)
+        elif op == "contains":
+            where_parts.append(f'"{col}" LIKE ?')
+            params.append(f"%{v}%")
+        elif op == "notcontains":
+            where_parts.append(f'"{col}" NOT LIKE ?')
+            params.append(f"%{v}%")
+
+    where_sql = " AND ".join(where_parts) if where_parts else "TRUE"
+    order_sql = ", ".join(order_parts) if order_parts else None
+    return where_sql, params, order_sql

tests/api/test_filters.py

@@ -0,0 +1,141 @@
+import polars as pl
+import pytest
+
+from src.api.filters import FilterError, build_where
+
+SCHEMA = pl.Schema(
+    {
+        "uid": pl.String,
+        "objet": pl.String,
+        "montant": pl.Float64,
+        "annee": pl.Int64,
+        "dateNotification": pl.Date,
+    }
+)
+
+
+def test_no_filters_returns_true():
+    where, params, order = build_where([], SCHEMA)
+    assert where == "TRUE"
+    assert params == []
+    assert order is None
+
+
+def test_exact_filter():
+    where, params, _ = build_where([("uid__exact", "abc")], SCHEMA)
+    assert where == '"uid" = ?'
+    assert params == ["abc"]
+
+
+def test_contains_filter_uses_like_wildcards():
+    where, params, _ = build_where([("objet__contains", "informatique")], SCHEMA)
+    assert where == '"objet" LIKE ?'
+    assert params == ["%informatique%"]
+
+
+def test_notcontains_filter():
+    where, params, _ = build_where([("objet__notcontains", "x")], SCHEMA)
+    assert where == '"objet" NOT LIKE ?'
+    assert params == ["%x%"]
+
+
+def test_comparison_operators_on_int():
+    where, params, _ = build_where([("annee__strictly_greater", "2023")], SCHEMA)
+    assert where == '"annee" > ?'
+    assert params == [2024 - 1]  # int coercion: 2023
+
+
+def test_in_filter_splits_on_commas():
+    where, params, _ = build_where([("annee__in", "2022,2023,2024")], SCHEMA)
+    assert where == '"annee" IN (?,?,?)'
+    assert params == [2022, 2023, 2024]
+
+
+def test_notin_filter():
+    where, params, _ = build_where([("annee__notin", "2020,2021")], SCHEMA)
+    assert where == '"annee" NOT IN (?,?)'
+    assert params == [2020, 2021]
+
+
+def test_isnull_filter_ignores_value():
+    where, params, _ = build_where([("dateNotification__isnull", "anything")], SCHEMA)
+    assert where == '"dateNotification" IS NULL'
+    assert params == []
+
+
+def test_isnotnull_filter():
+    where, params, _ = build_where([("dateNotification__isnotnull", "")], SCHEMA)
+    assert where == '"dateNotification" IS NOT NULL'
+
+
+def test_multiple_filters_joined_by_and():
+    where, params, _ = build_where(
+        [("uid__exact", "a"), ("annee__greater", "2020")], SCHEMA
+    )
+    assert where == '"uid" = ? AND "annee" >= ?'
+    assert params == ["a", 2020]
+
+
+def test_unknown_column_raises():
+    with pytest.raises(FilterError) as exc:
+        build_where([("foo__exact", "bar")], SCHEMA)
+    assert "foo" in str(exc.value)
+    assert exc.value.field == "foo__exact"
+
+
+def test_unknown_operator_raises():
+    with pytest.raises(FilterError) as exc:
+        build_where([("uid__weird", "x")], SCHEMA)
+    assert "weird" in str(exc.value)
+
+
+def test_bad_int_value_raises():
+    with pytest.raises(FilterError):
+        build_where([("annee__exact", "notanint")], SCHEMA)
+
+
+def test_bad_date_value_raises():
+    with pytest.raises(FilterError):
+        build_where([("dateNotification__exact", "notadate")], SCHEMA)
+
+
+def test_date_iso_coercion():
+    where, params, _ = build_where(
+        [("dateNotification__greater", "2024-01-01")], SCHEMA
+    )
+    from datetime import date
+
+    assert params == [date(2024, 1, 1)]
+
+
+def test_reserved_params_are_ignored():
+    where, params, order = build_where(
+        [
+            ("page", "2"),
+            ("page_size", "100"),
+            ("columns", "uid"),
+            ("count", "false"),
+            ("uid__exact", "z"),
+        ],
+        SCHEMA,
+    )
+    assert where == '"uid" = ?'
+    assert params == ["z"]
+
+
+def test_sort_returns_order_by():
+    where, params, order = build_where(
+        [("annee__sort", "desc"), ("uid__sort", "asc")], SCHEMA
+    )
+    assert where == "TRUE"
+    assert order == '"annee" DESC, "uid" ASC'
+
+
+def test_sort_invalid_direction_raises():
+    with pytest.raises(FilterError):
+        build_where([("uid__sort", "sideways")], SCHEMA)
+
+
+def test_param_without_operator_raises():
+    with pytest.raises(FilterError):
+        build_where([("uidexact", "x")], SCHEMA)