123 lines
4.2 KiB
Python
123 lines
4.2 KiB
Python
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))
|