feat: add support for SQLAlchemy func expressions in filter classes (#585)

Adds support for SQLAlchemy func() expressions in filter classes to eliminate type checker errors when using database functions like `func.random()` or `func.lower()`.

Author: cofin

.pre-commit-config.yaml

@@ -22,7 +22,7 @@ repos:
       - id: unasyncd
         additional_dependencies: ["ruff"]
   - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: "v0.14.0"
+    rev: "v0.14.2"
     hooks:
       # Run the linter.
       - id: ruff

advanced_alchemy/filters.py

@@ -26,6 +26,7 @@
 from dataclasses import dataclass
 from operator import attrgetter
 from typing import (
+    TYPE_CHECKING,
     Any,
     Callable,
     ClassVar,
@@ -53,13 +54,15 @@
     text,
     true,
 )
-from sqlalchemy.orm import InstrumentedAttribute
 from sqlalchemy.sql import operators as op
 from sqlalchemy.sql.dml import ReturningDelete, ReturningUpdate
 from typing_extensions import TypeAlias, TypedDict, TypeVar
 
 from advanced_alchemy.base import ModelProtocol
 
+if TYPE_CHECKING:
+    from sqlalchemy.orm import InstrumentedAttribute
+
 __all__ = (
     "BeforeAfter",
     "CollectionFilter",
@@ -153,7 +156,9 @@ def append_to_statement(
         return statement
 
     @staticmethod
-    def _get_instrumented_attr(model: Any, key: Union[str, InstrumentedAttribute[Any]]) -> InstrumentedAttribute[Any]:
+    def _get_instrumented_attr(
+        model: Any, key: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]"
+    ) -> "Union[ColumnElement[Any], InstrumentedAttribute[Any]]":
         """Get SQLAlchemy instrumented attribute from model.
 
         Args:
@@ -166,9 +171,7 @@ def _get_instrumented_attr(model: Any, key: Union[str, InstrumentedAttribute[Any
         See Also:
             :class:`sqlalchemy.orm.attributes.InstrumentedAttribute`: SQLAlchemy attribute
         """
-        if isinstance(key, str):
-            return cast("InstrumentedAttribute[Any]", getattr(model, key))
-        return key
+        return cast("InstrumentedAttribute[Any]", getattr(model, key)) if isinstance(key, str) else key
 
 
 @dataclass
@@ -186,8 +189,8 @@ class BeforeAfter(StatementFilter):
 
     """
 
-    field_name: str
-    """Name of the model attribute to filter on."""
+    field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]"
+    """Field name, model attribute, or func expression."""
     before: Optional[datetime.datetime]
     """Filter results where field is earlier than this value."""
     after: Optional[datetime.datetime]
@@ -232,8 +235,8 @@ class OnBeforeAfter(StatementFilter):
 
     """
 
-    field_name: str
-    """Name of the model attribute to filter on."""
+    field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]"
+    """Field name, model attribute, or func expression."""
     on_or_before: Optional[datetime.datetime]
     """Filter results where field is on or earlier than this value."""
     on_or_after: Optional[datetime.datetime]
@@ -280,8 +283,8 @@ class CollectionFilter(InAnyFilter, Generic[T]):
     Use ``prefer_any=True`` in ``append_to_statement`` to use the ``ANY`` operator.
     """
 
-    field_name: str
-    """Name of the model attribute to filter on."""
+    field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]"
+    """Field name, model attribute, or func expression."""
     values: Union[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)"""
@@ -339,8 +342,8 @@ class NotInCollectionFilter(InAnyFilter, Generic[T]):
 
     """
 
-    field_name: str
-    """Name of the model attribute to filter on."""
+    field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]"
+    """Field name, model attribute, or func expression."""
     values: Union[Collection[T], None]
     """Values for the ``NOT IN`` clause. If None or empty, no filter is applied."""
 
@@ -443,8 +446,8 @@ class OrderBy(StatementFilter):
         - :meth:`sqlalchemy.sql.expression.ColumnElement.desc`: Descending order
     """
 
-    field_name: str
-    """Name of the model attribute to sort on."""
+    field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]"
+    """Field name, model attribute, or func expression (e.g., ``func.random()``)."""
     sort_order: Literal["asc", "desc"] = "asc"
     """Sort direction ("asc" or "desc")."""
 
@@ -614,8 +617,8 @@ class ComparisonFilter(StatementFilter):
         ValueError: If an invalid operator is provided
     """
 
-    field_name: str
-    """Name of the model attribute to filter on."""
+    field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]"
+    """Field name, model attribute, or func expression."""
     operator: str
     """Comparison operator to use (one of 'eq', 'ne', 'gt', 'ge', 'lt', 'le')."""
     value: Any

advanced_alchemy/repository/memory/_async.py

@@ -209,36 +209,58 @@ def _exclude_unused_kwargs(self, kwargs: dict[str, Any]) -> dict[str, Any]:
     def _apply_limit_offset_pagination(result: list[ModelT], limit: int, offset: int) -> list[ModelT]:
         return result[offset:limit]
 
-    @staticmethod
+    def _extract_field_name(self, field: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]") -> str:
+        """Extract string field name from various input types.
+
+        Args:
+            field: Field name, column element, or instrumented attribute
+
+        Returns:
+            str: String field name for use with getattr()
+
+        Raises:
+            RepositoryError: If a ColumnElement (func expression) is used with mock repository
+        """
+        if isinstance(field, str):
+            return field
+        if isinstance(field, InstrumentedAttribute):
+            return field.key
+        msg = f"{type(field)} columns are not supported in mock repositories (in-memory filtering)"
+        raise RepositoryError(msg)
+
     def _filter_in_collection(
+        self,
         result: list[ModelT],
-        field_name: str,
+        field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]",
         values: abc.Collection[Any],
     ) -> list[ModelT]:
-        return [item for item in result if getattr(item, field_name) in values]
+        field_str = self._extract_field_name(field_name)
+        return [item for item in result if getattr(item, field_str) in values]
 
-    @staticmethod
     def _filter_not_in_collection(
+        self,
         result: list[ModelT],
-        field_name: str,
+        field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]",
         values: abc.Collection[Any],
     ) -> list[ModelT]:
         if not values:
             return result
-        return [item for item in result if getattr(item, field_name) not in values]
+        field_str = self._extract_field_name(field_name)
+        return [item for item in result if getattr(item, field_str) not in values]
 
-    @staticmethod
     def _filter_on_datetime_field(
+        self,
         result: list[ModelT],
-        field_name: str,
+        field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]",
         before: Optional[datetime.datetime] = None,
         after: Optional[datetime.datetime] = None,
         on_or_before: Optional[datetime.datetime] = None,
         on_or_after: Optional[datetime.datetime] = None,
     ) -> list[ModelT]:
+        field_str = self._extract_field_name(field_name)
         result_: list[ModelT] = []
         for item in result:
-            attr: datetime.datetime = getattr(item, field_name)
+            attr: datetime.datetime = getattr(item, field_str)
             if before is not None and attr < before:
                 result_.append(item)
             if after is not None and attr > after:
@@ -302,9 +324,13 @@ def _filter_result_by_kwargs(
         except AttributeError as error:
             raise RepositoryError from error
 
-    @staticmethod
-    def _order_by(result: list[ModelT], field_name: str, sort_desc: bool = False) -> list[ModelT]:
-        return sorted(result, key=lambda item: getattr(item, field_name), reverse=sort_desc)
+    def _order_by(
+        self,
+        result: list[ModelT],
+        field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]",
+        sort_desc: bool = False,
+    ) -> list[ModelT]:
+        return sorted(result, key=lambda item: getattr(item, self._extract_field_name(field_name)), reverse=sort_desc)
 
     def _apply_filters(
         self,

advanced_alchemy/repository/memory/_sync.py

@@ -210,36 +210,58 @@ def _exclude_unused_kwargs(self, kwargs: dict[str, Any]) -> dict[str, Any]:
     def _apply_limit_offset_pagination(result: list[ModelT], limit: int, offset: int) -> list[ModelT]:
         return result[offset:limit]
 
-    @staticmethod
+    def _extract_field_name(self, field: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]") -> str:
+        """Extract string field name from various input types.
+
+        Args:
+            field: Field name, column element, or instrumented attribute
+
+        Returns:
+            str: String field name for use with getattr()
+
+        Raises:
+            RepositoryError: If a ColumnElement (func expression) is used with mock repository
+        """
+        if isinstance(field, str):
+            return field
+        if isinstance(field, InstrumentedAttribute):
+            return field.key
+        msg = f"{type(field)} columns are not supported in mock repositories (in-memory filtering)"
+        raise RepositoryError(msg)
+
     def _filter_in_collection(
+        self,
         result: list[ModelT],
-        field_name: str,
+        field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]",
         values: abc.Collection[Any],
     ) -> list[ModelT]:
-        return [item for item in result if getattr(item, field_name) in values]
+        field_str = self._extract_field_name(field_name)
+        return [item for item in result if getattr(item, field_str) in values]
 
-    @staticmethod
     def _filter_not_in_collection(
+        self,
         result: list[ModelT],
-        field_name: str,
+        field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]",
         values: abc.Collection[Any],
     ) -> list[ModelT]:
         if not values:
             return result
-        return [item for item in result if getattr(item, field_name) not in values]
+        field_str = self._extract_field_name(field_name)
+        return [item for item in result if getattr(item, field_str) not in values]
 
-    @staticmethod
     def _filter_on_datetime_field(
+        self,
         result: list[ModelT],
-        field_name: str,
+        field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]",
         before: Optional[datetime.datetime] = None,
         after: Optional[datetime.datetime] = None,
         on_or_before: Optional[datetime.datetime] = None,
         on_or_after: Optional[datetime.datetime] = None,
     ) -> list[ModelT]:
+        field_str = self._extract_field_name(field_name)
         result_: list[ModelT] = []
         for item in result:
-            attr: datetime.datetime = getattr(item, field_name)
+            attr: datetime.datetime = getattr(item, field_str)
             if before is not None and attr < before:
                 result_.append(item)
             if after is not None and attr > after:
@@ -303,9 +325,13 @@ def _filter_result_by_kwargs(
         except AttributeError as error:
             raise RepositoryError from error
 
-    @staticmethod
-    def _order_by(result: list[ModelT], field_name: str, sort_desc: bool = False) -> list[ModelT]:
-        return sorted(result, key=lambda item: getattr(item, field_name), reverse=sort_desc)
+    def _order_by(
+        self,
+        result: list[ModelT],
+        field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]",
+        sort_desc: bool = False,
+    ) -> list[ModelT]:
+        return sorted(result, key=lambda item: getattr(item, self._extract_field_name(field_name)), reverse=sort_desc)
 
     def _apply_filters(
         self,

tests/integration/test_filters.py

@@ -3,7 +3,7 @@
 
 import pytest
 from pytest import FixtureRequest
-from sqlalchemy import Engine, String, select
+from sqlalchemy import Engine, String, func, select
 from sqlalchemy.ext.asyncio import AsyncEngine
 from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
 
@@ -456,6 +456,78 @@ def test_order_by_filter(session: Session, movie_model_sync: type[DeclarativeBas
     assert results[0].title == "The Hangover"
 
 
+def test_order_by_with_func_random(session: Session, movie_model_sync: type[DeclarativeBase]) -> None:
+    """Test OrderBy filter with func.random() expression."""
+    Movie = movie_model_sync
+
+    # Skip mock engines
+    dialect_name = getattr(session.bind.dialect, "name", "")
+    if dialect_name == "mock":
+        pytest.skip("Mock engines not supported for filter tests")
+
+    # Skip Oracle - uses dbms_random.value() instead of random()
+    if dialect_name.startswith("oracle"):
+        pytest.skip("Oracle uses dbms_random.value() instead of random()")
+
+    # Clean any existing data first, then setup fresh data
+    if dialect_name != "mock":
+        session.execute(Movie.__table__.delete())
+        session.commit()
+    setup_movie_data(session, Movie)
+
+    # Test func.random() - should not raise type error
+    order_by_filter = OrderBy(field_name=func.random())
+    statement = order_by_filter.append_to_statement(select(Movie), Movie)
+    results = session.execute(statement).scalars().all()
+    # Should return all movies, order is random
+    assert len(results) == 3
+
+
+def test_order_by_with_func_lower(session: Session, movie_model_sync: type[DeclarativeBase]) -> None:
+    """Test OrderBy filter with func.lower() for case-insensitive sorting."""
+    Movie = movie_model_sync
+
+    # Skip mock engines
+    if getattr(session.bind.dialect, "name", "") == "mock":
+        pytest.skip("Mock engines not supported for filter tests")
+
+    # Clean any existing data first, then setup fresh data
+    if getattr(session.bind.dialect, "name", "") != "mock":
+        session.execute(Movie.__table__.delete())
+        session.commit()
+    setup_movie_data(session, Movie)
+
+    # Test func.lower() for case-insensitive alphabetical sorting
+    order_by_filter = OrderBy(field_name=func.lower(Movie.title), sort_order="asc")
+    statement = order_by_filter.append_to_statement(select(Movie), Movie)
+    results = session.execute(statement).scalars().all()
+    # Should be sorted alphabetically: Shawshank, The Hangover, The Matrix
+    assert results[0].title == "Shawshank Redemption"
+    assert results[1].title == "The Hangover"
+    assert results[2].title == "The Matrix"
+
+
+def test_order_by_with_instrumented_attribute(session: Session, movie_model_sync: type[DeclarativeBase]) -> None:
+    """Test OrderBy filter with InstrumentedAttribute (Model.field)."""
+    Movie = movie_model_sync
+
+    # Skip mock engines
+    if getattr(session.bind.dialect, "name", "") == "mock":
+        pytest.skip("Mock engines not supported for filter tests")
+
+    # Clean any existing data first, then setup fresh data
+    if getattr(session.bind.dialect, "name", "") != "mock":
+        session.execute(Movie.__table__.delete())
+        session.commit()
+    setup_movie_data(session, Movie)
+
+    # Test with InstrumentedAttribute (backward compatibility)
+    order_by_filter = OrderBy(field_name=Movie.release_date, sort_order="asc")
+    statement = order_by_filter.append_to_statement(select(Movie), Movie)
+    results = session.execute(statement).scalars().all()
+    assert results[0].title == "Shawshank Redemption"
+
+
 def test_search_filter(session: Session, movie_model_sync: type[DeclarativeBase]) -> None:
     Movie = movie_model_sync