feat(query): filtermodel_to_ast (filterModel AG Grid -> AST) (#41)

src/utils/query_ast.py

@@ -146,3 +146,69 @@ def _numeric_to_sql(cond: Condition, col_type, quoted: str) -> tuple[str, list]:
         return f"{quoted} BETWEEN ? AND ?", [v1, v2]
     logger.warning(f"Opérateur numérique invalide : {cond.operator!r}")
     return "TRUE", []
+
+
+_TEXT_TYPE = {
+    "contains": "contains",
+    "notContains": "notContains",
+    "equals": "eq",
+    "notEqual": "neq",
+    "startsWith": "startsWith",
+    "endsWith": "endsWith",
+    "blank": "blank",
+    "notBlank": "notBlank",
+}
+_NUM_TYPE = {
+    "equals": "eq",
+    "notEqual": "neq",
+    "lessThan": "lt",
+    "lessThanOrEqual": "lte",
+    "greaterThan": "gt",
+    "greaterThanOrEqual": "gte",
+    "inRange": "range",
+    "blank": "blank",
+    "notBlank": "notBlank",
+}
+
+
+def _leaf(column: str, spec: dict):
+    """Convertit une condition AG Grid unitaire en Condition."""
+    ftype = spec.get("filterType", "text")
+    ag_type = spec.get("type")
+    if ftype == "date":
+        op = _NUM_TYPE.get(ag_type)
+        if op == "range":
+            return Condition(column, "range", spec.get("dateFrom"), spec.get("dateTo"))
+        return Condition(column, op, spec.get("dateFrom")) if op else None
+    if ftype == "number":
+        op = _NUM_TYPE.get(ag_type)
+        if op == "range":
+            return Condition(column, "range", spec.get("filter"), spec.get("filterTo"))
+        return Condition(column, op, spec.get("filter")) if op else None
+    # texte
+    op = _TEXT_TYPE.get(ag_type)
+    return Condition(column, op, spec.get("filter")) if op else None
+
+
+def filtermodel_to_ast(filter_model, schema):
+    """Traduit un filterModel AG Grid en AST. Colonnes combinées en And."""
+    if not filter_model:
+        return None
+    children = []
+    for column, spec in filter_model.items():
+        if column not in schema.names():
+            logger.warning(f"Filtre sur colonne inconnue ignoré : {column!r}")
+            continue
+        if "operator" in spec:  # deux conditions
+            c1 = _leaf(column, spec.get("condition1", {}))
+            c2 = _leaf(column, spec.get("condition2", {}))
+            parts = [c for c in (c1, c2) if c is not None]
+            if not parts:
+                continue
+            node = And(parts) if spec["operator"] == "AND" else Or(parts)
+        else:
+            node = _leaf(column, spec)
+            if node is None:
+                continue
+        children.append(node)
+    return And(children) if children else None

tests/test_query_ast.py

@@ -1,6 +1,6 @@
 import polars as pl
 
-from src.utils.query_ast import And, Condition, Not, Or, ast_to_sql
+from src.utils.query_ast import And, Condition, Not, Or, ast_to_sql, filtermodel_to_ast
 
 SCHEMA = pl.Schema(
     {
@@ -130,3 +130,71 @@ def test_and_or_not_grouping():
     sql, params = _run(node)
     assert " OR " in sql and " AND " in sql and "NOT (" in sql
     assert params == ["%beton%", "%ciment%", "%demolition%"]
+
+
+def test_filtermodel_empty_is_none():
+    assert filtermodel_to_ast(None, SCHEMA) is None
+    assert filtermodel_to_ast({}, SCHEMA) is None
+
+
+def test_filtermodel_text_contains():
+    fm = {"objet": {"filterType": "text", "type": "contains", "filter": "voirie"}}
+    _, params = ast_to_sql(filtermodel_to_ast(fm, SCHEMA), SCHEMA)
+    assert params == ["%voirie%"]
+
+
+def test_filtermodel_number_greaterthan():
+    fm = {"montant": {"filterType": "number", "type": "greaterThan", "filter": 40000}}
+    sql, params = ast_to_sql(filtermodel_to_ast(fm, SCHEMA), SCHEMA)
+    assert '"montant"' in sql and params == [40000.0]
+
+
+def test_filtermodel_number_inrange():
+    fm = {
+        "montant": {
+            "filterType": "number",
+            "type": "inRange",
+            "filter": 100,
+            "filterTo": 200,
+        }
+    }
+    _, params = ast_to_sql(filtermodel_to_ast(fm, SCHEMA), SCHEMA)
+    assert params == [100.0, 200.0]
+
+
+def test_filtermodel_date_uses_datefrom():
+    fm = {
+        "dateNotification": {
+            "filterType": "date",
+            "type": "greaterThan",
+            "dateFrom": "2022-01-01",
+        }
+    }
+    _, params = ast_to_sql(filtermodel_to_ast(fm, SCHEMA), SCHEMA)
+    assert params == ["2022-01-01"]
+
+
+def test_filtermodel_two_conditions_or():
+    fm = {
+        "objet": {
+            "filterType": "text",
+            "operator": "OR",
+            "condition1": {"filterType": "text", "type": "contains", "filter": "beton"},
+            "condition2": {
+                "filterType": "text",
+                "type": "contains",
+                "filter": "ciment",
+            },
+        }
+    }
+    sql, params = ast_to_sql(filtermodel_to_ast(fm, SCHEMA), SCHEMA)
+    assert " OR " in sql and params == ["%beton%", "%ciment%"]
+
+
+def test_filtermodel_multiple_columns_are_anded():
+    fm = {
+        "objet": {"filterType": "text", "type": "contains", "filter": "voirie"},
+        "montant": {"filterType": "number", "type": "greaterThan", "filter": 1000},
+    }
+    sql, params = ast_to_sql(filtermodel_to_ast(fm, SCHEMA), SCHEMA)
+    assert " AND " in sql and set(params) == {"%voirie%", 1000.0}