pyiceberg/expressions/__init__.py (540 lines of code) (raw):
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
from __future__ import annotations
from abc import ABC, abstractmethod
from functools import cached_property
from typing import (
Any,
Callable,
Generic,
Iterable,
Sequence,
Set,
Tuple,
Type,
TypeVar,
Union,
)
from pyiceberg.expressions.literals import (
AboveMax,
BelowMin,
Literal,
literal,
)
from pyiceberg.schema import Accessor, Schema
from pyiceberg.typedef import L, StructProtocol
from pyiceberg.types import DoubleType, FloatType, NestedField
from pyiceberg.utils.singleton import Singleton
def _to_unbound_term(term: Union[str, UnboundTerm[Any]]) -> UnboundTerm[Any]:
return Reference(term) if isinstance(term, str) else term
def _to_literal_set(values: Union[Iterable[L], Iterable[Literal[L]]]) -> Set[Literal[L]]:
return {_to_literal(v) for v in values}
def _to_literal(value: Union[L, Literal[L]]) -> Literal[L]:
if isinstance(value, Literal):
return value
else:
return literal(value)
class BooleanExpression(ABC):
"""An expression that evaluates to a boolean."""
@abstractmethod
def __invert__(self) -> BooleanExpression:
"""Transform the Expression into its negated version."""
def __and__(self, other: BooleanExpression) -> BooleanExpression:
"""Perform and operation on another expression."""
if not isinstance(other, BooleanExpression):
raise ValueError(f"Expected BooleanExpression, got: {other}")
return And(self, other)
def __or__(self, other: BooleanExpression) -> BooleanExpression:
"""Perform or operation on another expression."""
if not isinstance(other, BooleanExpression):
raise ValueError(f"Expected BooleanExpression, got: {other}")
return Or(self, other)
def _build_balanced_tree(
operator_: Callable[[BooleanExpression, BooleanExpression], BooleanExpression], items: Sequence[BooleanExpression]
) -> BooleanExpression:
"""
Recursively constructs a balanced binary tree of BooleanExpressions using the provided binary operator.
This function is a safer and more scalable alternative to:
reduce(operator_, items)
Using `reduce` creates a deeply nested, unbalanced tree (e.g., operator_(a, operator_(b, operator_(c, ...)))),
which grows linearly with the number of items. This can lead to RecursionError exceptions in Python
when the number of expressions is large (e.g., >1000).
In contrast, this function builds a balanced binary tree with logarithmic depth (O(log n)),
helping avoid recursion issues and ensuring that expression trees remain stable, predictable,
and safe to traverse — especially in tools like PyIceberg that operate on large logical trees.
Parameters:
operator_ (Callable): A binary operator function (e.g., pyiceberg.expressions.Or, And) that takes two
BooleanExpressions and returns a combined BooleanExpression.
items (Sequence[BooleanExpression]): A sequence of BooleanExpression objects to combine.
Returns:
BooleanExpression: The balanced combination of all input BooleanExpressions.
Raises:
ValueError: If the input sequence is empty.
"""
if not items:
raise ValueError("No expressions to combine")
if len(items) == 1:
return items[0]
mid = len(items) // 2
left = _build_balanced_tree(operator_, items[:mid])
right = _build_balanced_tree(operator_, items[mid:])
return operator_(left, right)
class Term(Generic[L], ABC):
"""A simple expression that evaluates to a value."""
class Bound(ABC):
"""Represents a bound value expression."""
B = TypeVar("B")
class Unbound(Generic[B], ABC):
"""Represents an unbound value expression."""
@abstractmethod
def bind(self, schema: Schema, case_sensitive: bool = True) -> B: ...
@property
@abstractmethod
def as_bound(self) -> Type[Bound]: ...
class BoundTerm(Term[L], Bound, ABC):
"""Represents a bound term."""
@abstractmethod
def ref(self) -> BoundReference[L]:
"""Return the bound reference."""
@abstractmethod
def eval(self, struct: StructProtocol) -> L: # pylint: disable=W0613
"""Return the value at the referenced field's position in an object that abides by the StructProtocol."""
class BoundReference(BoundTerm[L]):
"""A reference bound to a field in a schema.
Args:
field (NestedField): A referenced field in an Iceberg schema.
accessor (Accessor): An Accessor object to access the value at the field's position.
"""
field: NestedField
accessor: Accessor
def __init__(self, field: NestedField, accessor: Accessor):
self.field = field
self.accessor = accessor
def eval(self, struct: StructProtocol) -> L:
"""Return the value at the referenced field's position in an object that abides by the StructProtocol.
Args:
struct (StructProtocol): A row object that abides by the StructProtocol and returns values given a position.
Returns:
Any: The value at the referenced field's position in `struct`.
"""
return self.accessor.get(struct)
def __eq__(self, other: Any) -> bool:
"""Return the equality of two instances of the BoundReference class."""
return self.field == other.field if isinstance(other, BoundReference) else False
def __repr__(self) -> str:
"""Return the string representation of the BoundReference class."""
return f"BoundReference(field={repr(self.field)}, accessor={repr(self.accessor)})"
def ref(self) -> BoundReference[L]:
return self
def __hash__(self) -> int:
"""Return hash value of the BoundReference class."""
return hash(str(self))
class UnboundTerm(Term[Any], Unbound[BoundTerm[L]], ABC):
"""Represents an unbound term."""
@abstractmethod
def bind(self, schema: Schema, case_sensitive: bool = True) -> BoundTerm[L]: ...
class Reference(UnboundTerm[Any]):
"""A reference not yet bound to a field in a schema.
Args:
name (str): The name of the field.
Note:
An unbound reference is sometimes referred to as a "named" reference.
"""
name: str
def __init__(self, name: str) -> None:
self.name = name
def __repr__(self) -> str:
"""Return the string representation of the Reference class."""
return f"Reference(name={repr(self.name)})"
def __eq__(self, other: Any) -> bool:
"""Return the equality of two instances of the Reference class."""
return self.name == other.name if isinstance(other, Reference) else False
def bind(self, schema: Schema, case_sensitive: bool = True) -> BoundReference[L]:
"""Bind the reference to an Iceberg schema.
Args:
schema (Schema): An Iceberg schema.
case_sensitive (bool): Whether to consider case when binding the reference to the field.
Raises:
ValueError: If an empty name is provided.
Returns:
BoundReference: A reference bound to the specific field in the Iceberg schema.
"""
field = schema.find_field(name_or_id=self.name, case_sensitive=case_sensitive)
accessor = schema.accessor_for_field(field.field_id)
return self.as_bound(field=field, accessor=accessor) # type: ignore
@property
def as_bound(self) -> Type[BoundReference[L]]:
return BoundReference[L]
class And(BooleanExpression):
"""AND operation expression - logical conjunction."""
left: BooleanExpression
right: BooleanExpression
def __new__(cls, left: BooleanExpression, right: BooleanExpression, *rest: BooleanExpression) -> BooleanExpression: # type: ignore
if rest:
return _build_balanced_tree(And, (left, right, *rest))
if left is AlwaysFalse() or right is AlwaysFalse():
return AlwaysFalse()
elif left is AlwaysTrue():
return right
elif right is AlwaysTrue():
return left
else:
obj = super().__new__(cls)
obj.left = left
obj.right = right
return obj
def __eq__(self, other: Any) -> bool:
"""Return the equality of two instances of the And class."""
return self.left == other.left and self.right == other.right if isinstance(other, And) else False
def __str__(self) -> str:
"""Return the string representation of the And class."""
return f"And(left={str(self.left)}, right={str(self.right)})"
def __repr__(self) -> str:
"""Return the string representation of the And class."""
return f"And(left={repr(self.left)}, right={repr(self.right)})"
def __invert__(self) -> BooleanExpression:
"""Transform the Expression into its negated version."""
# De Morgan's law: not (A and B) = (not A) or (not B)
return Or(~self.left, ~self.right)
def __getnewargs__(self) -> Tuple[BooleanExpression, BooleanExpression]:
"""Pickle the And class."""
return (self.left, self.right)
class Or(BooleanExpression):
"""OR operation expression - logical disjunction."""
left: BooleanExpression
right: BooleanExpression
def __new__(cls, left: BooleanExpression, right: BooleanExpression, *rest: BooleanExpression) -> BooleanExpression: # type: ignore
if rest:
return _build_balanced_tree(Or, (left, right, *rest))
if left is AlwaysTrue() or right is AlwaysTrue():
return AlwaysTrue()
elif left is AlwaysFalse():
return right
elif right is AlwaysFalse():
return left
else:
obj = super().__new__(cls)
obj.left = left
obj.right = right
return obj
def __eq__(self, other: Any) -> bool:
"""Return the equality of two instances of the Or class."""
return self.left == other.left and self.right == other.right if isinstance(other, Or) else False
def __repr__(self) -> str:
"""Return the string representation of the Or class."""
return f"Or(left={repr(self.left)}, right={repr(self.right)})"
def __invert__(self) -> BooleanExpression:
"""Transform the Expression into its negated version."""
# De Morgan's law: not (A or B) = (not A) and (not B)
return And(~self.left, ~self.right)
def __getnewargs__(self) -> Tuple[BooleanExpression, BooleanExpression]:
"""Pickle the Or class."""
return (self.left, self.right)
class Not(BooleanExpression):
"""NOT operation expression - logical negation."""
child: BooleanExpression
def __new__(cls, child: BooleanExpression) -> BooleanExpression: # type: ignore
if child is AlwaysTrue():
return AlwaysFalse()
elif child is AlwaysFalse():
return AlwaysTrue()
elif isinstance(child, Not):
return child.child
obj = super().__new__(cls)
obj.child = child
return obj
def __repr__(self) -> str:
"""Return the string representation of the Not class."""
return f"Not(child={repr(self.child)})"
def __eq__(self, other: Any) -> bool:
"""Return the equality of two instances of the Not class."""
return self.child == other.child if isinstance(other, Not) else False
def __invert__(self) -> BooleanExpression:
"""Transform the Expression into its negated version."""
return self.child
def __getnewargs__(self) -> Tuple[BooleanExpression]:
"""Pickle the Not class."""
return (self.child,)
class AlwaysTrue(BooleanExpression, Singleton):
"""TRUE expression."""
def __invert__(self) -> AlwaysFalse:
"""Transform the Expression into its negated version."""
return AlwaysFalse()
def __str__(self) -> str:
"""Return the string representation of the AlwaysTrue class."""
return "AlwaysTrue()"
def __repr__(self) -> str:
"""Return the string representation of the AlwaysTrue class."""
return "AlwaysTrue()"
class AlwaysFalse(BooleanExpression, Singleton):
"""FALSE expression."""
def __invert__(self) -> AlwaysTrue:
"""Transform the Expression into its negated version."""
return AlwaysTrue()
def __str__(self) -> str:
"""Return the string representation of the AlwaysFalse class."""
return "AlwaysFalse()"
def __repr__(self) -> str:
"""Return the string representation of the AlwaysFalse class."""
return "AlwaysFalse()"
class BoundPredicate(Generic[L], Bound, BooleanExpression, ABC):
term: BoundTerm[L]
def __init__(self, term: BoundTerm[L]):
self.term = term
def __eq__(self, other: Any) -> bool:
"""Return the equality of two instances of the BoundPredicate class."""
if isinstance(other, self.__class__):
return self.term == other.term
return False
@property
@abstractmethod
def as_unbound(self) -> Type[UnboundPredicate[Any]]: ...
class UnboundPredicate(Generic[L], Unbound[BooleanExpression], BooleanExpression, ABC):
term: UnboundTerm[Any]
def __init__(self, term: Union[str, UnboundTerm[Any]]):
self.term = _to_unbound_term(term)
def __eq__(self, other: Any) -> bool:
"""Return the equality of two instances of the UnboundPredicate class."""
return self.term == other.term if isinstance(other, self.__class__) else False
@abstractmethod
def bind(self, schema: Schema, case_sensitive: bool = True) -> BooleanExpression: ...
@property
@abstractmethod
def as_bound(self) -> Type[BoundPredicate[L]]: ...
class UnaryPredicate(UnboundPredicate[Any], ABC):
def bind(self, schema: Schema, case_sensitive: bool = True) -> BoundUnaryPredicate[Any]:
bound_term = self.term.bind(schema, case_sensitive)
return self.as_bound(bound_term)
def __repr__(self) -> str:
"""Return the string representation of the UnaryPredicate class."""
return f"{str(self.__class__.__name__)}(term={repr(self.term)})"
@property
@abstractmethod
def as_bound(self) -> Type[BoundUnaryPredicate[Any]]: ...
class BoundUnaryPredicate(BoundPredicate[L], ABC):
def __repr__(self) -> str:
"""Return the string representation of the BoundUnaryPredicate class."""
return f"{str(self.__class__.__name__)}(term={repr(self.term)})"
@property
@abstractmethod
def as_unbound(self) -> Type[UnaryPredicate]: ...
def __getnewargs__(self) -> Tuple[BoundTerm[L]]:
"""Pickle the BoundUnaryPredicate class."""
return (self.term,)
class BoundIsNull(BoundUnaryPredicate[L]):
def __new__(cls, term: BoundTerm[L]) -> BooleanExpression: # type: ignore # pylint: disable=W0221
if term.ref().field.required:
return AlwaysFalse()
return super().__new__(cls)
def __invert__(self) -> BoundNotNull[L]:
"""Transform the Expression into its negated version."""
return BoundNotNull(self.term)
@property
def as_unbound(self) -> Type[IsNull]:
return IsNull
class BoundNotNull(BoundUnaryPredicate[L]):
def __new__(cls, term: BoundTerm[L]): # type: ignore # pylint: disable=W0221
if term.ref().field.required:
return AlwaysTrue()
return super().__new__(cls)
def __invert__(self) -> BoundIsNull[L]:
"""Transform the Expression into its negated version."""
return BoundIsNull(self.term)
@property
def as_unbound(self) -> Type[NotNull]:
return NotNull
class IsNull(UnaryPredicate):
def __invert__(self) -> NotNull:
"""Transform the Expression into its negated version."""
return NotNull(self.term)
@property
def as_bound(self) -> Type[BoundIsNull[L]]:
return BoundIsNull[L]
class NotNull(UnaryPredicate):
def __invert__(self) -> IsNull:
"""Transform the Expression into its negated version."""
return IsNull(self.term)
@property
def as_bound(self) -> Type[BoundNotNull[L]]:
return BoundNotNull[L]
class BoundIsNaN(BoundUnaryPredicate[L]):
def __new__(cls, term: BoundTerm[L]) -> BooleanExpression: # type: ignore # pylint: disable=W0221
bound_type = term.ref().field.field_type
if isinstance(bound_type, (FloatType, DoubleType)):
return super().__new__(cls)
return AlwaysFalse()
def __invert__(self) -> BoundNotNaN[L]:
"""Transform the Expression into its negated version."""
return BoundNotNaN(self.term)
@property
def as_unbound(self) -> Type[IsNaN]:
return IsNaN
class BoundNotNaN(BoundUnaryPredicate[L]):
def __new__(cls, term: BoundTerm[L]) -> BooleanExpression: # type: ignore # pylint: disable=W0221
bound_type = term.ref().field.field_type
if isinstance(bound_type, (FloatType, DoubleType)):
return super().__new__(cls)
return AlwaysTrue()
def __invert__(self) -> BoundIsNaN[L]:
"""Transform the Expression into its negated version."""
return BoundIsNaN(self.term)
@property
def as_unbound(self) -> Type[NotNaN]:
return NotNaN
class IsNaN(UnaryPredicate):
def __invert__(self) -> NotNaN:
"""Transform the Expression into its negated version."""
return NotNaN(self.term)
@property
def as_bound(self) -> Type[BoundIsNaN[L]]:
return BoundIsNaN[L]
class NotNaN(UnaryPredicate):
def __invert__(self) -> IsNaN:
"""Transform the Expression into its negated version."""
return IsNaN(self.term)
@property
def as_bound(self) -> Type[BoundNotNaN[L]]:
return BoundNotNaN[L]
class SetPredicate(UnboundPredicate[L], ABC):
literals: Set[Literal[L]]
def __init__(self, term: Union[str, UnboundTerm[Any]], literals: Union[Iterable[L], Iterable[Literal[L]]]):
super().__init__(term)
self.literals = _to_literal_set(literals)
def bind(self, schema: Schema, case_sensitive: bool = True) -> BoundSetPredicate[L]:
bound_term = self.term.bind(schema, case_sensitive)
return self.as_bound(bound_term, {lit.to(bound_term.ref().field.field_type) for lit in self.literals})
def __str__(self) -> str:
"""Return the string representation of the SetPredicate class."""
# Sort to make it deterministic
return f"{str(self.__class__.__name__)}({str(self.term)}, {{{', '.join(sorted([str(literal) for literal in self.literals]))}}})"
def __repr__(self) -> str:
"""Return the string representation of the SetPredicate class."""
# Sort to make it deterministic
return f"{str(self.__class__.__name__)}({repr(self.term)}, {{{', '.join(sorted([repr(literal) for literal in self.literals]))}}})"
def __eq__(self, other: Any) -> bool:
"""Return the equality of two instances of the SetPredicate class."""
return self.term == other.term and self.literals == other.literals if isinstance(other, self.__class__) else False
def __getnewargs__(self) -> Tuple[UnboundTerm[L], Set[Literal[L]]]:
"""Pickle the SetPredicate class."""
return (self.term, self.literals)
@property
@abstractmethod
def as_bound(self) -> Type[BoundSetPredicate[L]]:
return BoundSetPredicate[L]
class BoundSetPredicate(BoundPredicate[L], ABC):
literals: Set[Literal[L]]
def __init__(self, term: BoundTerm[L], literals: Set[Literal[L]]):
# Since we don't know the type of BoundPredicate[L], we have to ignore this one
super().__init__(term) # type: ignore
self.literals = _to_literal_set(literals) # pylint: disable=W0621
@cached_property
def value_set(self) -> Set[L]:
return {lit.value for lit in self.literals}
def __str__(self) -> str:
"""Return the string representation of the BoundSetPredicate class."""
# Sort to make it deterministic
return f"{str(self.__class__.__name__)}({str(self.term)}, {{{', '.join(sorted([str(literal) for literal in self.literals]))}}})"
def __repr__(self) -> str:
"""Return the string representation of the BoundSetPredicate class."""
# Sort to make it deterministic
return f"{str(self.__class__.__name__)}({repr(self.term)}, {{{', '.join(sorted([repr(literal) for literal in self.literals]))}}})"
def __eq__(self, other: Any) -> bool:
"""Return the equality of two instances of the BoundSetPredicate class."""
return self.term == other.term and self.literals == other.literals if isinstance(other, self.__class__) else False
def __getnewargs__(self) -> Tuple[BoundTerm[L], Set[Literal[L]]]:
"""Pickle the BoundSetPredicate class."""
return (self.term, self.literals)
@property
@abstractmethod
def as_unbound(self) -> Type[SetPredicate[L]]: ...
class BoundIn(BoundSetPredicate[L]):
def __new__(cls, term: BoundTerm[L], literals: Set[Literal[L]]) -> BooleanExpression: # type: ignore # pylint: disable=W0221
count = len(literals)
if count == 0:
return AlwaysFalse()
elif count == 1:
return BoundEqualTo(term, next(iter(literals)))
else:
return super().__new__(cls)
def __invert__(self) -> BoundNotIn[L]:
"""Transform the Expression into its negated version."""
return BoundNotIn(self.term, self.literals)
def __eq__(self, other: Any) -> bool:
"""Return the equality of two instances of the BoundIn class."""
return self.term == other.term and self.literals == other.literals if isinstance(other, self.__class__) else False
@property
def as_unbound(self) -> Type[In[L]]:
return In
class BoundNotIn(BoundSetPredicate[L]):
def __new__( # type: ignore # pylint: disable=W0221
cls,
term: BoundTerm[L],
literals: Set[Literal[L]],
) -> BooleanExpression:
count = len(literals)
if count == 0:
return AlwaysTrue()
elif count == 1:
return BoundNotEqualTo(term, next(iter(literals)))
else:
return super().__new__(cls)
def __invert__(self) -> BoundIn[L]:
"""Transform the Expression into its negated version."""
return BoundIn(self.term, self.literals)
@property
def as_unbound(self) -> Type[NotIn[L]]:
return NotIn
class In(SetPredicate[L]):
def __new__( # type: ignore # pylint: disable=W0221
cls, term: Union[str, UnboundTerm[Any]], literals: Union[Iterable[L], Iterable[Literal[L]]]
) -> BooleanExpression:
literals_set: Set[Literal[L]] = _to_literal_set(literals)
count = len(literals_set)
if count == 0:
return AlwaysFalse()
elif count == 1:
return EqualTo(term, next(iter(literals))) # type: ignore
else:
return super().__new__(cls)
def __invert__(self) -> NotIn[L]:
"""Transform the Expression into its negated version."""
return NotIn[L](self.term, self.literals)
@property
def as_bound(self) -> Type[BoundIn[L]]:
return BoundIn[L]
class NotIn(SetPredicate[L], ABC):
def __new__( # type: ignore # pylint: disable=W0221
cls, term: Union[str, UnboundTerm[Any]], literals: Union[Iterable[L], Iterable[Literal[L]]]
) -> BooleanExpression:
literals_set: Set[Literal[L]] = _to_literal_set(literals)
count = len(literals_set)
if count == 0:
return AlwaysTrue()
elif count == 1:
return NotEqualTo(term, next(iter(literals_set)))
else:
return super().__new__(cls)
def __invert__(self) -> In[L]:
"""Transform the Expression into its negated version."""
return In[L](self.term, self.literals)
@property
def as_bound(self) -> Type[BoundNotIn[L]]:
return BoundNotIn[L]
class LiteralPredicate(UnboundPredicate[L], ABC):
literal: Literal[L]
def __init__(self, term: Union[str, UnboundTerm[Any]], literal: Union[L, Literal[L]]): # pylint: disable=W0621
super().__init__(term)
self.literal = _to_literal(literal) # pylint: disable=W0621
def bind(self, schema: Schema, case_sensitive: bool = True) -> BoundLiteralPredicate[L]:
bound_term = self.term.bind(schema, case_sensitive)
lit = self.literal.to(bound_term.ref().field.field_type)
if isinstance(lit, AboveMax):
if isinstance(self, (LessThan, LessThanOrEqual, NotEqualTo)):
return AlwaysTrue() # type: ignore
elif isinstance(self, (GreaterThan, GreaterThanOrEqual, EqualTo)):
return AlwaysFalse() # type: ignore
elif isinstance(lit, BelowMin):
if isinstance(self, (GreaterThan, GreaterThanOrEqual, NotEqualTo)):
return AlwaysTrue() # type: ignore
elif isinstance(self, (LessThan, LessThanOrEqual, EqualTo)):
return AlwaysFalse() # type: ignore
return self.as_bound(bound_term, lit)
def __eq__(self, other: Any) -> bool:
"""Return the equality of two instances of the LiteralPredicate class."""
if isinstance(other, self.__class__):
return self.term == other.term and self.literal == other.literal
return False
def __repr__(self) -> str:
"""Return the string representation of the LiteralPredicate class."""
return f"{str(self.__class__.__name__)}(term={repr(self.term)}, literal={repr(self.literal)})"
@property
@abstractmethod
def as_bound(self) -> Type[BoundLiteralPredicate[L]]: ...
class BoundLiteralPredicate(BoundPredicate[L], ABC):
literal: Literal[L]
def __init__(self, term: BoundTerm[L], literal: Literal[L]): # pylint: disable=W0621
# Since we don't know the type of BoundPredicate[L], we have to ignore this one
super().__init__(term) # type: ignore
self.literal = literal # pylint: disable=W0621
def __eq__(self, other: Any) -> bool:
"""Return the equality of two instances of the BoundLiteralPredicate class."""
if isinstance(other, self.__class__):
return self.term == other.term and self.literal == other.literal
return False
def __repr__(self) -> str:
"""Return the string representation of the BoundLiteralPredicate class."""
return f"{str(self.__class__.__name__)}(term={repr(self.term)}, literal={repr(self.literal)})"
@property
@abstractmethod
def as_unbound(self) -> Type[LiteralPredicate[L]]: ...
class BoundEqualTo(BoundLiteralPredicate[L]):
def __invert__(self) -> BoundNotEqualTo[L]:
"""Transform the Expression into its negated version."""
return BoundNotEqualTo[L](self.term, self.literal)
@property
def as_unbound(self) -> Type[EqualTo[L]]:
return EqualTo
class BoundNotEqualTo(BoundLiteralPredicate[L]):
def __invert__(self) -> BoundEqualTo[L]:
"""Transform the Expression into its negated version."""
return BoundEqualTo[L](self.term, self.literal)
@property
def as_unbound(self) -> Type[NotEqualTo[L]]:
return NotEqualTo
class BoundGreaterThanOrEqual(BoundLiteralPredicate[L]):
def __invert__(self) -> BoundLessThan[L]:
"""Transform the Expression into its negated version."""
return BoundLessThan[L](self.term, self.literal)
@property
def as_unbound(self) -> Type[GreaterThanOrEqual[L]]:
return GreaterThanOrEqual[L]
class BoundGreaterThan(BoundLiteralPredicate[L]):
def __invert__(self) -> BoundLessThanOrEqual[L]:
"""Transform the Expression into its negated version."""
return BoundLessThanOrEqual(self.term, self.literal)
@property
def as_unbound(self) -> Type[GreaterThan[L]]:
return GreaterThan[L]
class BoundLessThan(BoundLiteralPredicate[L]):
def __invert__(self) -> BoundGreaterThanOrEqual[L]:
"""Transform the Expression into its negated version."""
return BoundGreaterThanOrEqual[L](self.term, self.literal)
@property
def as_unbound(self) -> Type[LessThan[L]]:
return LessThan[L]
class BoundLessThanOrEqual(BoundLiteralPredicate[L]):
def __invert__(self) -> BoundGreaterThan[L]:
"""Transform the Expression into its negated version."""
return BoundGreaterThan[L](self.term, self.literal)
@property
def as_unbound(self) -> Type[LessThanOrEqual[L]]:
return LessThanOrEqual[L]
class BoundStartsWith(BoundLiteralPredicate[L]):
def __invert__(self) -> BoundNotStartsWith[L]:
"""Transform the Expression into its negated version."""
return BoundNotStartsWith[L](self.term, self.literal)
@property
def as_unbound(self) -> Type[StartsWith[L]]:
return StartsWith[L]
class BoundNotStartsWith(BoundLiteralPredicate[L]):
def __invert__(self) -> BoundStartsWith[L]:
"""Transform the Expression into its negated version."""
return BoundStartsWith[L](self.term, self.literal)
@property
def as_unbound(self) -> Type[NotStartsWith[L]]:
return NotStartsWith[L]
class EqualTo(LiteralPredicate[L]):
def __invert__(self) -> NotEqualTo[L]:
"""Transform the Expression into its negated version."""
return NotEqualTo[L](self.term, self.literal)
@property
def as_bound(self) -> Type[BoundEqualTo[L]]:
return BoundEqualTo[L]
class NotEqualTo(LiteralPredicate[L]):
def __invert__(self) -> EqualTo[L]:
"""Transform the Expression into its negated version."""
return EqualTo[L](self.term, self.literal)
@property
def as_bound(self) -> Type[BoundNotEqualTo[L]]:
return BoundNotEqualTo[L]
class LessThan(LiteralPredicate[L]):
def __invert__(self) -> GreaterThanOrEqual[L]:
"""Transform the Expression into its negated version."""
return GreaterThanOrEqual[L](self.term, self.literal)
@property
def as_bound(self) -> Type[BoundLessThan[L]]:
return BoundLessThan[L]
class GreaterThanOrEqual(LiteralPredicate[L]):
def __invert__(self) -> LessThan[L]:
"""Transform the Expression into its negated version."""
return LessThan[L](self.term, self.literal)
@property
def as_bound(self) -> Type[BoundGreaterThanOrEqual[L]]:
return BoundGreaterThanOrEqual[L]
class GreaterThan(LiteralPredicate[L]):
def __invert__(self) -> LessThanOrEqual[L]:
"""Transform the Expression into its negated version."""
return LessThanOrEqual[L](self.term, self.literal)
@property
def as_bound(self) -> Type[BoundGreaterThan[L]]:
return BoundGreaterThan[L]
class LessThanOrEqual(LiteralPredicate[L]):
def __invert__(self) -> GreaterThan[L]:
"""Transform the Expression into its negated version."""
return GreaterThan[L](self.term, self.literal)
@property
def as_bound(self) -> Type[BoundLessThanOrEqual[L]]:
return BoundLessThanOrEqual[L]
class StartsWith(LiteralPredicate[L]):
def __invert__(self) -> NotStartsWith[L]:
"""Transform the Expression into its negated version."""
return NotStartsWith[L](self.term, self.literal)
@property
def as_bound(self) -> Type[BoundStartsWith[L]]:
return BoundStartsWith[L]
class NotStartsWith(LiteralPredicate[L]):
def __invert__(self) -> StartsWith[L]:
"""Transform the Expression into its negated version."""
return StartsWith[L](self.term, self.literal)
@property
def as_bound(self) -> Type[BoundNotStartsWith[L]]:
return BoundNotStartsWith[L]