Initial 2
This commit is contained in:
122
example/filters.py
Normal file
122
example/filters.py
Normal file
@ -0,0 +1,122 @@
|
||||
class FieldOperator(ABC):
|
||||
model_cls: Type[Base]
|
||||
|
||||
def __init__(self, model_cls: Type[Base]) -> None:
|
||||
self.model_cls = model_cls
|
||||
|
||||
@abstractmethod
|
||||
def apply_query(self, field: str, value: Any, query: Query) -> Query: ...
|
||||
|
||||
def to_query(self, field: str, value: Any, query: Query) -> Query:
|
||||
return self.apply_query(field, value, query)
|
||||
|
||||
|
||||
class EQ(FieldOperator):
|
||||
"""ColumnOperators.__eq__() (Python “==” operator)"""
|
||||
|
||||
def apply_query(self, field: str, value: Any, query: Query) -> Query:
|
||||
return query.where(getattr(self.model_cls, field) == value)
|
||||
|
||||
|
||||
class NOT_EQ(FieldOperator):
|
||||
"""ColumnOperators.__ne__() (Python “!=” operator)"""
|
||||
|
||||
def apply_query(self, field: str, value: Any, query: Query) -> Query:
|
||||
return query.where(getattr(self.model_cls, field) != value)
|
||||
|
||||
|
||||
class GT(FieldOperator):
|
||||
"""ColumnOperators.__gt__() (Python “>” operator)"""
|
||||
|
||||
def apply_query(self, field: str, value: Comparable, query: Query) -> Query:
|
||||
return query.where(getattr(self.model_cls, field) > value)
|
||||
|
||||
|
||||
class GE(FieldOperator):
|
||||
"""ColumnOperators.__ge__() (Python “>=” operator)"""
|
||||
|
||||
def apply_query(self, field: str, value: Comparable, query: Query) -> Query:
|
||||
return query.where(getattr(self.model_cls, field) >= value)
|
||||
|
||||
|
||||
class LT(FieldOperator):
|
||||
"""ColumnOperators.__lt__() (Python “<” operator)"""
|
||||
|
||||
def apply_query(self, field: str, value: Comparable, query: Query) -> Query:
|
||||
return query.where(getattr(self.model_cls, field) < value)
|
||||
|
||||
|
||||
class LE(FieldOperator):
|
||||
"""ColumnOperators.__le__() (Python “<=” operator)"""
|
||||
|
||||
def apply_query(self, field: str, value: Comparable, query: Query) -> Query:
|
||||
return query.where(getattr(self.model_cls, field) <= value)
|
||||
|
||||
|
||||
class IN(FieldOperator):
|
||||
"""IN is available most typically by passing a list of values to the ColumnOperators.in_() method"""
|
||||
|
||||
def apply_query(self, field: str, value: Iterable[Any], query: Query) -> Query:
|
||||
return query.where(getattr(self.model_cls, field).in_(value))
|
||||
|
||||
|
||||
class NOT_IN(FieldOperator):
|
||||
"""NOT IN is available via the ColumnOperators.not_in() operator"""
|
||||
|
||||
def apply_query(self, field: str, value: Iterable[Any], query: Query) -> Query:
|
||||
return query.where(~getattr(self.model_cls, field).in_(value))
|
||||
|
||||
|
||||
class LIKE(FieldOperator):
|
||||
"""ColumnOperators.like(). Expects string value."""
|
||||
|
||||
def apply_query(self, field: str, value: str, query: Query) -> Query:
|
||||
return query.where(getattr(self.model_cls, field).like(f"%{value}%"))
|
||||
|
||||
|
||||
class ILIKE(FieldOperator):
|
||||
"""ColumnOperators.ilike(). Expects string value."""
|
||||
|
||||
def apply_query(self, field: str, value: str, query: Query) -> Query:
|
||||
return query.where(getattr(self.model_cls, field).ilike(f"%{value}%"))
|
||||
|
||||
|
||||
class IS_NULL(FieldOperator):
|
||||
"""Test value for NULL / NOT NULL"""
|
||||
|
||||
def apply_query(self, field: str, value: bool, query: Query) -> Query:
|
||||
if value:
|
||||
return query.where(getattr(self.model_cls, field).is_(None))
|
||||
else:
|
||||
return query.where(getattr(self.model_cls, field).is_not(None))
|
||||
|
||||
|
||||
class IS_NOT_NULL(FieldOperator):
|
||||
"""Test value for NULL / NOT NULL"""
|
||||
|
||||
def apply_query(self, field: str, value: bool, query: Query) -> Query:
|
||||
if value:
|
||||
return query.where(getattr(self.model_cls, field).is_not(None))
|
||||
else:
|
||||
return query.where(getattr(self.model_cls, field).is_(None))
|
||||
|
||||
|
||||
class IEXACT(FieldOperator):
|
||||
"""ColumnOperators.__eq__() over strings but comparison is case-insensitive. Expects string value."""
|
||||
|
||||
def apply_query(self, field: str, value: str, query: Query) -> Query:
|
||||
return query.where(func.lower(getattr(self.model_cls, field)) == value.lower())
|
||||
|
||||
|
||||
class REGEX(FieldOperator):
|
||||
"""Case-sensitive regex operator. Expects string regex pattern."""
|
||||
|
||||
def apply_query(self, field: str, value: str, query: Query) -> Query:
|
||||
return query.where(getattr(self.model_cls, field).op("~")(value))
|
||||
|
||||
|
||||
class IREGEX(FieldOperator):
|
||||
"""Case-insensitive regex operator. Expects string regex pattern."""
|
||||
|
||||
def apply_query(self, field: str, value: str, query: Query) -> Query:
|
||||
return query.where(getattr(self.model_cls, field).op("~*")(value))
|
||||
Reference in New Issue
Block a user