diff --git a/src/utils/filter.py b/src/utils/filter.py index 8df3590c..f7510797 100644 --- a/src/utils/filter.py +++ b/src/utils/filter.py @@ -5,6 +5,7 @@ from typing import Any, TypeVar from typing import cast as typing_cast from sqlalchemy import ColumnElement, Select, and_, case, cast, literal, not_, or_ +from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.types import Numeric from ..exceptions import FilterError @@ -345,6 +346,14 @@ def _build_field_condition( # For JSONB fields (metadata, configuration), check if it contains nested comparison operators if column_name in JSONB_COLUMNS: return _build_nested_metadata_conditions(column, value) # pyright: ignore + elif not isinstance(column.type, JSONB): + # A dict against a scalar column compiles fine but fails in the + # driver at execute time ("cannot adapt type 'dict'") as a 500. + # Reject unknown operator dicts here as a 422 instead. + keys = sorted(typing_cast("dict[str, Any]", value)) + raise FilterError( + f"Invalid filter for column '{key}': unsupported operator(s) {keys}. Expected one of {sorted(COMPARISON_OPERATORS)} or a scalar value." + ) else: return column == value else: diff --git a/tests/utils/test_filter.py b/tests/utils/test_filter.py new file mode 100644 index 00000000..49f84e23 --- /dev/null +++ b/tests/utils/test_filter.py @@ -0,0 +1,31 @@ +"""Unit tests for filter condition building.""" + +import pytest +from sqlalchemy import select + +from src.exceptions import FilterError +from src.models import Document +from src.utils.filter import apply_filter + + +def test_unknown_operator_dict_on_scalar_column_raises(): + """An unrecognized operator dict must 422, not reach the driver as a 500. + + Regression: {"session_id": {"operator": "null"}} compiled to + `session_name = %(param)s` with a dict bind, which psycopg rejected with + "cannot adapt type 'dict'" -> unhandled 500. + """ + with pytest.raises(FilterError): + apply_filter(select(Document), Document, {"session_id": {"operator": "null"}}) + + +def test_known_operator_dict_on_scalar_column_still_works(): + stmt = apply_filter( + select(Document), Document, {"session_id": {"in": ["s1", "s2"]}} + ) + assert "session_name IN" in str(stmt).replace("documents.", "") + + +def test_dict_on_jsonb_column_still_works(): + stmt = apply_filter(select(Document), Document, {"metadata": {"kind": "note"}}) + assert "internal_metadata" in str(stmt)