From a3cfad3d8fcc3cd1cb91f6bb6c248cdd7561142f Mon Sep 17 00:00:00 2001 From: Colin Maudry Date: Wed, 13 May 2026 13:45:55 +0200 Subject: [PATCH] =?UTF-8?q?API:=20parser=20de=20filtres=20et=20g=C3=A9n?= =?UTF-8?q?=C3=A9ration=20SQL=20param=C3=A9tr=C3=A9=20(#78)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- src/api/filters.py | 138 +++++++++++++++++++++++++++++++++++++ tests/api/test_filters.py | 141 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 279 insertions(+) create mode 100644 src/api/filters.py create mode 100644 tests/api/test_filters.py diff --git a/src/api/filters.py b/src/api/filters.py new file mode 100644 index 0000000..573dde8 --- /dev/null +++ b/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 diff --git a/tests/api/test_filters.py b/tests/api/test_filters.py new file mode 100644 index 0000000..b7706fe --- /dev/null +++ b/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)