feat(filters): adds an ExistsFilter and NotExists filter

advanced_alchemy/filters.py

@@ -36,8 +36,21 @@
 from dataclasses import dataclass
 from operator import attrgetter
 from typing import TYPE_CHECKING, Any, Callable, Generic, Literal, Optional, Union, cast
-
-from sqlalchemy import BinaryExpression, ColumnElement, Delete, Select, Update, and_, any_, or_, text
+
+from sqlalchemy import (
+    BinaryExpression,
+    Delete,
+    Select,
+    Update,
+    and_,
+    any_,
+    exists,
+    false,
+    not_,
+    or_,
+    select,
+    text,
+)
 from sqlalchemy.orm import InstrumentedAttribute
 from typing_extensions import TypeAlias, TypeVar
 
@@ -53,9 +66,11 @@
 __all__ = (
     "BeforeAfter",
     "CollectionFilter",
+    "ExistsFilter",
     "FilterTypes",
     "InAnyFilter",
     "LimitOffset",
+    "NotExistsFilter",
     "NotInCollectionFilter",
     "NotInSearchFilter",
     "OnBeforeAfter",
@@ -74,7 +89,7 @@
     "StatementTypeT",
     bound="Union[ReturningDelete[tuple[Any]], ReturningUpdate[tuple[Any]], Select[tuple[Any]], Select[Any], Update, Delete]",
 )
-FilterTypes: TypeAlias = "Union[BeforeAfter, OnBeforeAfter, CollectionFilter[Any], LimitOffset, OrderBy, SearchFilter, NotInCollectionFilter[Any], NotInSearchFilter]"
+FilterTypes: TypeAlias = "Union[BeforeAfter, OnBeforeAfter, CollectionFilter[Any], LimitOffset, OrderBy, SearchFilter, NotInCollectionFilter[Any], NotInSearchFilter | ExistsFilter | NotExistsFilter]"
 """Aggregate type alias of the types supported for collection filtering."""
 
 
@@ -571,3 +586,233 @@ def _func(self) -> "attrgetter[Callable[[str], BinaryExpression[bool]]]":
             - :meth:`sqlalchemy.sql.expression.ColumnOperators.notilike`: NOT ILIKE
         """
         return attrgetter("not_ilike" if self.ignore_case else "not_like")
+
+
+@dataclass
+class ExistsFilter(StatementFilter):
+    """Filter for EXISTS subqueries.
+
+    This filter creates an EXISTS condition using a list of column expressions.
+    The expressions can be combined using either AND or OR logic.
+
+    Args:
+        field_name: The field to narrow against in the EXISTS clause
+        values: List of SQLAlchemy column expressions to use in the EXISTS clause
+        operator: If "and", combines conditions with AND, otherwise uses OR. Defaults to "and".
+
+    Example:
+        Basic usage with AND conditions::
+
+            from sqlalchemy import select
+            from advanced_alchemy.filters import ExistsFilter
+
+            filter = ExistsFilter(
+                field_name="User.is_active",
+                values=[User.email.like("%@example.com%")],
+            )
+            statement = filter.append_to_statement(
+                select(Organization), Organization
+            )
+
+        Using OR conditions::
+
+            filter = ExistsFilter(
+                field_name="User.role",
+                values=[User.role == "admin", User.role == "owner"],
+                operator="or",
+            )
+    """
+
+    field_name: str
+    """Name of model attribute to search on."""
+    values: list[ColumnElement[bool]]
+    """List of SQLAlchemy column expressions to use in the EXISTS clause."""
+    operator: Literal["and", "or"] = "and"
+    """If "and", combines conditions with AND, otherwise uses OR."""
+
+    @property
+    def _and(self) -> Callable[..., ColumnElement[bool]]:
+        """Return the SQL operator for combining multiple search clauses.
+
+        Returns:
+            Callable[..., ColumnElement[bool]]: The `and_` operator for AND conditions
+
+        See Also:
+            :func:`sqlalchemy.sql.expression.and_`: SQLAlchemy AND operator
+        """
+        return and_
+
+    @property
+    def _or(self) -> Callable[..., ColumnElement[bool]]:
+        """Return the SQL operator for combining multiple search clauses.
+
+        Returns:
+            Callable[..., ColumnElement[bool]]: The `or_` operator for OR conditions
+
+        See Also:
+            :func:`sqlalchemy.sql.expression.or_`: SQLAlchemy OR operator
+        """
+        return or_
+
+    def get_exists_clause(self, model: type[ModelT]) -> ColumnElement[bool]:
+        """Generate the EXISTS clause for the statement.
+
+        Args:
+            model: The SQLAlchemy model class
+
+        Returns:
+            ColumnElement[bool]: EXISTS clause
+        """
+        field = self._get_instrumented_attr(model, self.field_name)
+
+        # Get the underlying column name of the field
+        field_column = getattr(field, "comparator", None)
+        if not field_column:
+            return false()  # Handle cases where the field might not be directly comparable, ie. relations
+        field_column_name = field_column.key
+
+        # Construct a subquery using select()
+        subquery = select(field).where(
+            *(
+                [getattr(model, field_column_name) == getattr(model, field_column_name), self._and(*self.values)]
+                if self.operator == "and"
+                else [getattr(model, field_column_name) == getattr(model, field_column_name), self._or(*self.values)]
+            )
+        )
+
+        # Use the subquery in the exists() clause
+        return exists(subquery)
+
+    def append_to_statement(self, statement: StatementTypeT, model: type[ModelT]) -> StatementTypeT:
+        """Apply EXISTS condition to the statement.
+
+        Args:
+            statement: The SQLAlchemy statement to modify
+            model: The SQLAlchemy model class
+
+        Returns:
+            StatementTypeT: Modified statement with EXISTS condition
+
+        Note:
+            The conditions are combined using AND or OR based on the operator parameter.
+        """
+        if not self.values:
+            return statement
+        exists_clause = self.get_exists_clause(model)
+        return cast("StatementTypeT", statement.where(exists_clause))
+
+
+@dataclass
+class NotExistsFilter(StatementFilter):
+    """Filter for NOT EXISTS subqueries.
+
+    This filter creates a NOT EXISTS condition using a list of column expressions.
+    The expressions can be combined using either AND or OR logic.
+
+    Args:
+        field_name: The field to narrow against in the NOT EXISTS clause
+        values: List of SQLAlchemy column expressions to use in the NOT EXISTS clause
+        operator: If "and", combines conditions with AND, otherwise uses OR. Defaults to "and".
+
+    Example:
+        Basic usage with AND conditions::
+
+            from sqlalchemy import select
+            from advanced_alchemy.filters import NotExistsFilter
+
+            filter = NotExistsFilter(
+                field_name="User.is_active",
+                values=[User.email.like("%@example.com%")],
+            )
+            statement = filter.append_to_statement(
+                select(Organization), Organization
+            )
+
+        Using OR conditions::
+
+            filter = NotExistsFilter(
+                field_name="User.role",
+                values=[User.role == "admin", User.role == "owner"],
+                operator="or",
+            )
+    """
+
+    field_name: str
+    """Name of model attribute to search on."""
+    values: list[ColumnElement[bool]]
+    """List of SQLAlchemy column expressions to use in the NOT EXISTS clause."""
+    operator: Literal["and", "or"] = "and"
+    """If "and", combines conditions with AND, otherwise uses OR."""
+
+    @property
+    def _and(self) -> Callable[..., ColumnElement[bool]]:
+        """Return the SQL operator for combining multiple search clauses.
+
+        Returns:
+            Callable[..., ColumnElement[bool]]: The `and_` operator for AND conditions
+
+        See Also:
+            :func:`sqlalchemy.sql.expression.and_`: SQLAlchemy AND operator
+        """
+        return and_
+
+    @property
+    def _or(self) -> Callable[..., ColumnElement[bool]]:
+        """Return the SQL operator for combining multiple search clauses.
+
+        Returns:
+            Callable[..., ColumnElement[bool]]: The `or_` operator for OR conditions
+
+        See Also:
+            :func:`sqlalchemy.sql.expression.or_`: SQLAlchemy OR operator
+        """
+        return or_
+
+    def get_exists_clause(self, model: type[ModelT]) -> ColumnElement[bool]:
+        """Generate the NOT EXISTS clause for the statement.
+
+        Args:
+            model: The SQLAlchemy model class
+
+        Returns:
+            ColumnElement[bool]: NOT EXISTS clause
+        """
+        field = self._get_instrumented_attr(model, self.field_name)
+
+        # Get the underlying column name of the field
+        field_column = getattr(field, "comparator", None)
+        if not field_column:
+            return false()  # Handle cases where the field might not be directly comparable, ie. relations
+        field_column_name = field_column.key
+
+        # Construct a subquery using select()
+        subquery = select(field).where(
+            *(
+                [getattr(model, field_column_name) == getattr(model, field_column_name), self._and(*self.values)]
+                if self.operator == "and"
+                else [
+                    getattr(model, field_column_name) == getattr(model, field_column_name),
+                    self._or(*self.values),
+                ]
+            )
+        )
+        # Use the subquery in the exists() clause and negate it with not_()
+        return not_(exists(subquery))
+
+    def append_to_statement(self, statement: StatementTypeT, model: type[ModelT]) -> StatementTypeT:
+        """Apply NOT EXISTS condition to the statement.
+
+        Args:
+            statement: The SQLAlchemy statement to modify
+            model: The SQLAlchemy model class
+
+        Returns:
+            StatementTypeT: Modified statement with NOT EXISTS condition
+
+        Note:
+            The conditions are combined using AND or OR based on the operator parameter.
+        """
+        if not self.values:
+            return statement
+        exists_clause = self.get_exists_clause(model)
+        return cast("StatementTypeT", statement.where(exists_clause))