1
- from typing import cast
1
+ from typing import (
2
+ TYPE_CHECKING ,
3
+ cast ,
4
+ )
2
5
3
- from sqlalchemy import func
6
+ from sqlalchemy import (
7
+ BinaryExpression ,
8
+ BooleanClauseList ,
9
+ func ,
10
+ )
4
11
from sqlalchemy .orm import InstrumentedAttribute
5
- from sqlalchemy import BinaryExpression
6
12
7
13
from fastapi_jsonapi .types_metadata import CustomFilterSQL
8
- from pydantic .fields import FieldInfo
9
14
15
+ if TYPE_CHECKING :
16
+ from pydantic .fields import FieldInfo
10
17
11
- class LowerEqualsFilterSQL (CustomFilterSQL [InstrumentedAttribute , BinaryExpression ]):
18
+
19
+ SQLAExpressionType = BinaryExpression | BooleanClauseList
20
+
21
+
22
+ class CustomFilterSQLA (CustomFilterSQL [InstrumentedAttribute , SQLAExpressionType ]):
23
+ """Base class for custom SQLAlchemy filters"""
24
+
25
+
26
+ class LowerEqualsFilterSQL (CustomFilterSQLA ):
12
27
def get_expression (
13
28
self ,
14
- schema_field : FieldInfo ,
29
+ schema_field : " FieldInfo" ,
15
30
model_column : InstrumentedAttribute ,
16
31
value : str ,
17
32
operator : str ,
@@ -22,4 +37,17 @@ def get_expression(
22
37
)
23
38
24
39
40
+ # TODO: tests coverage
41
+ class JSONBContainsFilterSQL (CustomFilterSQLA ):
42
+ def get_expression (
43
+ self ,
44
+ schema_field : "FieldInfo" ,
45
+ model_column : InstrumentedAttribute ,
46
+ value : str ,
47
+ operator : str ,
48
+ ) -> BinaryExpression :
49
+ return model_column .op ("@>" )(value )
50
+
51
+
25
52
sql_filter_lower_equals = LowerEqualsFilterSQL (op = "lower_equals" )
53
+ sql_filter_jsonb_contains = JSONBContainsFilterSQL (op = "jsonb_contains" )
0 commit comments