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))