diff --git a/src/utils/query_ast.py b/src/utils/query_ast.py index 802ba14..d81e8f1 100644 --- a/src/utils/query_ast.py +++ b/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 diff --git a/tests/test_query_ast.py b/tests/test_query_ast.py index 7fd0de7..fee6f2c 100644 --- a/tests/test_query_ast.py +++ b/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}