"""SQLAlchemy filter constructs for advanced query operations.
This module provides a comprehensive collection of filter datastructures designed to
enhance SQLAlchemy query construction. It implements type-safe, reusable filter patterns
for common database query operations.
Features:
Type-safe filter construction, datetime range filtering, collection-based filtering,
pagination support, search operations, and customizable ordering.
Example:
Basic usage with a datetime filter::
import datetime
from advanced_alchemy.filters import BeforeAfter
filter = BeforeAfter(
field_name="created_at",
before=datetime.datetime.now(),
after=datetime.datetime(2023, 1, 1),
)
statement = filter.append_to_statement(select(Model), Model)
Note:
All filter classes implement the :class:`StatementFilter` ABC, ensuring consistent
interface across different filter types.
See Also:
- :class:`sqlalchemy.sql.expression.Select`: Core SQLAlchemy select expression
- :class:`sqlalchemy.orm.Query`: SQLAlchemy ORM query interface
- :mod:`advanced_alchemy.base`: Base model definitions
"""
from __future__ import annotations
import datetime # noqa: TC003
from abc import ABC, abstractmethod
from collections import abc # noqa: TC003
from dataclasses import dataclass
from operator import attrgetter
from typing import TYPE_CHECKING, Any, Generic, Literal, cast
from sqlalchemy import BinaryExpression, Delete, Select, Update, and_, any_, or_, text
from typing_extensions import TypeVar
if TYPE_CHECKING:
from typing import Callable
from sqlalchemy import ColumnElement
from sqlalchemy.orm import InstrumentedAttribute
from sqlalchemy.sql.dml import ReturningDelete, ReturningUpdate
from typing_extensions import TypeAlias
from advanced_alchemy import base
__all__ = (
"BeforeAfter",
"CollectionFilter",
"FilterTypes",
"InAnyFilter",
"LimitOffset",
"NotInCollectionFilter",
"NotInSearchFilter",
"OnBeforeAfter",
"OrderBy",
"PaginationFilter",
"SearchFilter",
"StatementFilter",
"StatementFilterT",
"StatementTypeT",
)
T = TypeVar("T")
ModelT = TypeVar("ModelT", bound="base.ModelProtocol")
StatementFilterT = TypeVar("StatementFilterT", bound="StatementFilter")
StatementTypeT = TypeVar(
"StatementTypeT",
bound="ReturningDelete[tuple[Any]] | ReturningUpdate[tuple[Any]] | Select[tuple[Any]] | Select[Any] | Update | Delete",
)
FilterTypes: TypeAlias = "BeforeAfter | OnBeforeAfter | CollectionFilter[Any] | LimitOffset | OrderBy | SearchFilter | NotInCollectionFilter[Any] | NotInSearchFilter"
"""Aggregate type alias of the types supported for collection filtering."""
[docs]
class StatementFilter(ABC):
"""Abstract base class for SQLAlchemy statement filters.
This class defines the interface for all filter types in the system. Each filter
implementation must provide a method to append its filtering logic to an existing
SQLAlchemy statement.
"""
[docs]
@abstractmethod
def append_to_statement(
self, statement: StatementTypeT, model: type[ModelT], *args: Any, **kwargs: Any
) -> StatementTypeT:
"""Append filter conditions to a SQLAlchemy statement.
Args:
statement: The SQLAlchemy statement to modify
model: The SQLAlchemy model class
*args: Additional positional arguments
**kwargs: Additional keyword arguments
Returns:
StatementTypeT: Modified SQLAlchemy statement with filter conditions applied
Raises:
NotImplementedError: If the concrete class doesn't implement this method
Note:
This method must be implemented by all concrete filter classes.
See Also:
:meth:`sqlalchemy.sql.expression.Select.where`: SQLAlchemy where clause
"""
return statement
@staticmethod
def _get_instrumented_attr(model: Any, key: str | InstrumentedAttribute[Any]) -> InstrumentedAttribute[Any]:
"""Get SQLAlchemy instrumented attribute from model.
Args:
model: SQLAlchemy model class or instance
key: Attribute name or instrumented attribute
Returns:
InstrumentedAttribute[Any]: SQLAlchemy instrumented attribute
See Also:
:class:`sqlalchemy.orm.attributes.InstrumentedAttribute`: SQLAlchemy attribute
"""
if isinstance(key, str):
return cast("InstrumentedAttribute[Any]", getattr(model, key))
return key
[docs]
@dataclass
class BeforeAfter(StatementFilter):
"""DateTime range filter with exclusive bounds.
This filter creates date/time range conditions using < and > operators,
excluding the boundary values.
If either `before` or `after` is None, that boundary condition is not applied.
See Also:
---------
:class:`OnBeforeAfter` : Inclusive datetime range filtering
"""
field_name: str
"""Name of the model attribute to filter on."""
before: datetime.datetime | None
"""Filter results where field is earlier than this value."""
after: datetime.datetime | None
"""Filter results where field is later than this value."""
[docs]
def append_to_statement(self, statement: StatementTypeT, model: type[ModelT]) -> StatementTypeT:
"""Apply datetime range conditions to statement.
Parameters
----------
statement : StatementTypeT
The SQLAlchemy statement to modify
model : type[ModelT]
The SQLAlchemy model class
Returns:
--------
StatementTypeT
Modified statement with datetime range conditions
"""
field = self._get_instrumented_attr(model, self.field_name)
if self.before is not None:
statement = cast("StatementTypeT", statement.where(field < self.before))
if self.after is not None:
statement = cast("StatementTypeT", statement.where(field > self.after))
return statement
[docs]
@dataclass
class OnBeforeAfter(StatementFilter):
"""DateTime range filter with inclusive bounds.
This filter creates date/time range conditions using <= and >= operators,
including the boundary values.
If either `on_or_before` or `on_or_after` is None, that boundary condition
is not applied.
See Also:
---------
:class:`BeforeAfter` : Exclusive datetime range filtering
"""
field_name: str
"""Name of the model attribute to filter on."""
on_or_before: datetime.datetime | None
"""Filter results where field is on or earlier than this value."""
on_or_after: datetime.datetime | None
"""Filter results where field is on or later than this value."""
[docs]
def append_to_statement(self, statement: StatementTypeT, model: type[ModelT]) -> StatementTypeT:
"""Apply inclusive datetime range conditions to statement.
Parameters
----------
statement : StatementTypeT
The SQLAlchemy statement to modify
model : type[ModelT]
The SQLAlchemy model class
Returns:
--------
StatementTypeT
Modified statement with inclusive datetime range conditions
"""
field = self._get_instrumented_attr(model, self.field_name)
if self.on_or_before is not None:
statement = cast("StatementTypeT", statement.where(field <= self.on_or_before))
if self.on_or_after is not None:
statement = cast("StatementTypeT", statement.where(field >= self.on_or_after))
return statement
[docs]
class InAnyFilter(StatementFilter, ABC):
"""Base class for filters using IN or ANY operators.
This abstract class provides common functionality for filters that check
membership in a collection using either the SQL IN operator or the ANY operator.
"""
[docs]
@dataclass
class CollectionFilter(InAnyFilter, Generic[T]):
"""Data required to construct a WHERE ... IN (...) clause.
This filter restricts records based on a field's presence in a collection of values.
The filter supports both ``IN`` and ``ANY`` operators for collection membership testing.
Use ``prefer_any=True`` in ``append_to_statement`` to use the ``ANY`` operator.
"""
field_name: str
"""Name of the model attribute to filter on."""
values: abc.Collection[T] | None
"""Values for the ``IN`` clause. If this is None, no filter is applied.
An empty list will force an empty result set (WHERE 1=-1)"""
[docs]
def append_to_statement(
self,
statement: StatementTypeT,
model: type[ModelT],
prefer_any: bool = False,
) -> StatementTypeT:
"""Apply a WHERE ... IN or WHERE ... ANY (...) clause to the statement.
Parameters
----------
statement : StatementTypeT
The SQLAlchemy statement to modify
model : type[ModelT]
The SQLAlchemy model class
prefer_any : bool, optional
If True, uses the SQLAlchemy :func:`any_` operator instead of
:func:`in_` for the filter condition
Returns:
--------
StatementTypeT
Modified statement with the appropriate IN conditions
"""
field = self._get_instrumented_attr(model, self.field_name)
if self.values is None:
return statement
if not self.values:
# Return empty result set by forcing a false condition
return cast("StatementTypeT", statement.where(text("1=-1")))
if prefer_any:
return cast("StatementTypeT", statement.where(any_(self.values) == field)) # type: ignore[arg-type]
return cast("StatementTypeT", statement.where(field.in_(self.values)))
[docs]
@dataclass
class NotInCollectionFilter(InAnyFilter, Generic[T]):
"""Data required to construct a WHERE ... NOT IN (...) clause.
This filter restricts records based on a field's absence in a collection of values.
The filter supports both ``NOT IN`` and ``!= ANY`` operators for collection exclusion.
Use ``prefer_any=True`` in ``append_to_statement`` to use the ``ANY`` operator.
Parameters
----------
field_name : str
Name of the model attribute to filter on
values : abc.Collection[T] | None
Values for the ``NOT IN`` clause. If this is None or empty,
the filter is not applied.
"""
field_name: str
"""Name of the model attribute to filter on."""
values: abc.Collection[T] | None
"""Values for the ``NOT IN`` clause. If None or empty, no filter is applied."""
[docs]
def append_to_statement(
self,
statement: StatementTypeT,
model: type[ModelT],
prefer_any: bool = False,
) -> StatementTypeT:
"""Apply a WHERE ... NOT IN or WHERE ... != ANY(...) clause to the statement.
Parameters
----------
statement : StatementTypeT
The SQLAlchemy statement to modify
model : type[ModelT]
The SQLAlchemy model class
prefer_any : bool, optional
If True, uses the SQLAlchemy :func:`any_` operator instead of
:func:`notin_` for the filter condition
Returns:
--------
StatementTypeT
Modified statement with the appropriate NOT IN conditions
"""
field = self._get_instrumented_attr(model, self.field_name)
if not self.values:
# If None or empty, we do not modify the statement
return statement
if prefer_any:
return cast("StatementTypeT", statement.where(any_(self.values) != field)) # type: ignore[arg-type]
return cast("StatementTypeT", statement.where(field.notin_(self.values)))
[docs]
@dataclass
class LimitOffset(PaginationFilter):
"""Limit and offset pagination filter.
Implements traditional pagination using SQL LIMIT and OFFSET clauses.
Only applies to SELECT statements; other statement types are returned unmodified.
Note:
This filter only modifies SELECT statements. For other statement types
(UPDATE, DELETE), the statement is returned unchanged.
See Also:
- :meth:`sqlalchemy.sql.expression.Select.limit`: SQLAlchemy LIMIT clause
- :meth:`sqlalchemy.sql.expression.Select.offset`: SQLAlchemy OFFSET clause
"""
limit: int
"""Maximum number of rows to return."""
offset: int
"""Number of rows to skip before returning results."""
[docs]
def append_to_statement(self, statement: StatementTypeT, model: type[ModelT]) -> StatementTypeT:
"""Apply LIMIT/OFFSET pagination to the statement.
Args:
statement: The SQLAlchemy statement to modify
model: The SQLAlchemy model class
Returns:
StatementTypeT: Modified statement with limit and offset applied
Note:
Only modifies SELECT statements. Other statement types are returned as-is.
See Also:
:class:`sqlalchemy.sql.expression.Select`: SQLAlchemy SELECT statement
"""
if isinstance(statement, Select):
return cast("StatementTypeT", statement.limit(self.limit).offset(self.offset))
return statement
[docs]
@dataclass
class OrderBy(StatementFilter):
"""Order by a specific field.
Appends an ORDER BY clause to SELECT statements, sorting records by the
specified field in ascending or descending order.
Note:
This filter only modifies SELECT statements. For other statement types,
the statement is returned unchanged.
See Also:
- :meth:`sqlalchemy.sql.expression.Select.order_by`: SQLAlchemy ORDER BY clause
- :meth:`sqlalchemy.sql.expression.ColumnElement.asc`: Ascending order
- :meth:`sqlalchemy.sql.expression.ColumnElement.desc`: Descending order
"""
field_name: str
"""Name of the model attribute to sort on."""
sort_order: Literal["asc", "desc"] = "asc"
"""Sort direction ("asc" or "desc")."""
[docs]
def append_to_statement(self, statement: StatementTypeT, model: type[ModelT]) -> StatementTypeT:
"""Append an ORDER BY clause to the statement.
Args:
statement: The SQLAlchemy statement to modify
model: The SQLAlchemy model class
Returns:
StatementTypeT: Modified statement with an ORDER BY clause
Note:
Only modifies SELECT statements. Other statement types are returned as-is.
See Also:
:meth:`sqlalchemy.sql.expression.Select.order_by`: SQLAlchemy ORDER BY
"""
if not isinstance(statement, Select):
return statement
field = self._get_instrumented_attr(model, self.field_name)
if self.sort_order == "desc":
return cast("StatementTypeT", statement.order_by(field.desc()))
return cast("StatementTypeT", statement.order_by(field.asc()))
[docs]
@dataclass
class SearchFilter(StatementFilter):
"""Case-sensitive or case-insensitive substring matching filter.
Implements text search using SQL LIKE or ILIKE operators. Can search across
multiple fields using OR conditions.
Note:
The search pattern automatically adds wildcards before and after the search
value, equivalent to SQL pattern '%value%'.
See Also:
- :class:`.NotInSearchFilter`: Opposite filter using NOT LIKE/ILIKE
- :meth:`sqlalchemy.sql.expression.ColumnOperators.like`: Case-sensitive LIKE
- :meth:`sqlalchemy.sql.expression.ColumnOperators.ilike`: Case-insensitive LIKE
"""
field_name: str | set[str]
"""Name or set of names of model attributes to search on."""
value: str
"""Text to match within the field(s)."""
ignore_case: bool | None = False
"""Whether to use case-insensitive matching."""
@property
def _operator(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_
@property
def _func(self) -> attrgetter[Callable[[str], BinaryExpression[bool]]]:
"""Return the appropriate LIKE or ILIKE operator as a function.
Returns:
attrgetter: Bound method for LIKE or ILIKE operations
See Also:
- :meth:`sqlalchemy.sql.expression.ColumnOperators.like`: LIKE operator
- :meth:`sqlalchemy.sql.expression.ColumnOperators.ilike`: ILIKE operator
"""
return attrgetter("ilike" if self.ignore_case else "like")
@property
def normalized_field_names(self) -> set[str]:
"""Convert field_name to a set if it's a single string.
Returns:
set[str]: Set of field names to be searched
"""
return {self.field_name} if isinstance(self.field_name, str) else self.field_name
[docs]
def get_search_clauses(self, model: type[ModelT]) -> list[BinaryExpression[bool]]:
"""Generate the LIKE/ILIKE clauses for all specified fields.
Args:
model: The SQLAlchemy model class
Returns:
list[BinaryExpression[bool]]: List of text matching expressions
See Also:
:class:`sqlalchemy.sql.expression.BinaryExpression`: SQLAlchemy expression
"""
search_clause: list[BinaryExpression[bool]] = []
for field_name in self.normalized_field_names:
field = self._get_instrumented_attr(model, field_name)
search_text = f"%{self.value}%"
search_clause.append(self._func(field)(search_text))
return search_clause
[docs]
def append_to_statement(self, statement: StatementTypeT, model: type[ModelT]) -> StatementTypeT:
"""Append a LIKE/ILIKE clause to the statement.
Args:
statement: The SQLAlchemy statement to modify
model: The SQLAlchemy model class
Returns:
StatementTypeT: Modified statement with text search clauses
See Also:
:meth:`sqlalchemy.sql.expression.Select.where`: SQLAlchemy WHERE clause
"""
where_clause = self._operator(*self.get_search_clauses(model))
return cast("StatementTypeT", statement.where(where_clause))
[docs]
@dataclass
class NotInSearchFilter(SearchFilter):
"""Filter for excluding records that match a substring.
Implements negative text search using SQL NOT LIKE or NOT ILIKE operators.
Can exclude across multiple fields using AND conditions.
Args:
field_name: Name or set of names of model attributes to search on
value: Text to exclude from the field(s)
ignore_case: If True, uses NOT ILIKE for case-insensitive matching
Note:
Uses AND for multiple fields, meaning records matching any field will be excluded.
See Also:
- :class:`.SearchFilter`: Opposite filter using LIKE/ILIKE
- :meth:`sqlalchemy.sql.expression.ColumnOperators.notlike`: NOT LIKE operator
- :meth:`sqlalchemy.sql.expression.ColumnOperators.notilike`: NOT ILIKE operator
"""
@property
def _operator(self) -> Callable[..., ColumnElement[bool]]:
"""Return the SQL operator for combining multiple negated 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 _func(self) -> attrgetter[Callable[[str], BinaryExpression[bool]]]:
"""Return the appropriate NOT LIKE or NOT ILIKE operator as a function.
Returns:
attrgetter: Bound method for NOT LIKE or NOT ILIKE operations
See Also:
- :meth:`sqlalchemy.sql.expression.ColumnOperators.notlike`: NOT LIKE
- :meth:`sqlalchemy.sql.expression.ColumnOperators.notilike`: NOT ILIKE
"""
return attrgetter("not_ilike" if self.ignore_case else "not_like")
