fix(litestar): normalize sort filter defaults

Author: cofin

advanced_alchemy/extensions/fastapi/providers.py

@@ -48,7 +48,13 @@
     SQLAlchemyAsyncRepositoryService,
     SQLAlchemySyncRepositoryService,
 )
-from advanced_alchemy.utils.dependencies import DependencyCache, FieldNameType, FilterConfig, make_hashable
+from advanced_alchemy.utils.dependencies import (
+    DependencyCache,
+    FieldNameType,
+    FilterConfig,
+    make_hashable,
+    normalize_sort_field,
+)
 from advanced_alchemy.utils.text import camelize
 
 logger = logging.getLogger("advanced_alchemy.extensions.fastapi")
@@ -599,6 +605,7 @@ def provide_search_filter(
 
     # Add sort filter providers
     if sort_field := config.get("sort_field"):
+        sort_field_default = normalize_sort_field(sort_field)
         sort_order_default = config.get("sort_order", "desc")
 
         def provide_order_by(
@@ -609,7 +616,7 @@ def provide_order_by(
                     description="Field to order by.",
                     required=False,
                 ),
-            ] = sort_field,  # type: ignore[assignment]
+            ] = sort_field_default,
             sort_order: Annotated[
                 Optional[SortOrder],
                 Query(

advanced_alchemy/extensions/litestar/providers.py

@@ -43,7 +43,13 @@
     SQLAlchemyAsyncRepositoryService,
     SQLAlchemySyncRepositoryService,
 )
-from advanced_alchemy.utils.dependencies import DependencyCache, FieldNameType, FilterConfig, make_hashable
+from advanced_alchemy.utils.dependencies import (
+    DependencyCache,
+    FieldNameType,
+    FilterConfig,
+    make_hashable,
+    normalize_sort_field,
+)
 from advanced_alchemy.utils.singleton import SingletonMeta
 from advanced_alchemy.utils.text import camelize
 
@@ -317,6 +323,33 @@ def create_filter_dependencies(
     return deps
 
 
+def _create_order_by_filter_provider(
+    sort_field: Union[str, set[str], list[str]],
+    sort_order_default: SortOrder = "desc",
+) -> Callable[..., OrderBy]:
+    sort_field_default = normalize_sort_field(sort_field)
+
+    def provide_order_by(
+        field_name: Annotated[
+            StringOrNone,
+            QueryParameter(
+                title="Order by field",
+                name="orderBy",
+            ),
+        ] = sort_field_default,
+        sort_order: Annotated[
+            SortOrderOrNone,
+            QueryParameter(
+                title="Field to search",
+                name="sortOrder",
+            ),
+        ] = sort_order_default,
+    ) -> OrderBy:
+        return OrderBy(field_name=field_name, sort_order=sort_order or sort_order_default)  # type: ignore[arg-type]
+
+    return provide_order_by
+
+
 def _create_statement_filters(  # noqa: C901
     config: FilterConfig, dep_defaults: DependencyDefaults = DEPENDENCY_DEFAULTS
 ) -> dict[str, Provide]:
@@ -408,26 +441,10 @@ def provide_search_filter(
         filters[dep_defaults.SEARCH_FILTER_DEPENDENCY_KEY] = Provide(provide_search_filter, sync_to_thread=False)
 
     if sort_field := config.get("sort_field"):
-
-        def provide_order_by(
-            field_name: Annotated[
-                StringOrNone,
-                QueryParameter(
-                    title="Order by field",
-                    name="orderBy",
-                ),
-            ] = sort_field,  # type: ignore[assignment]
-            sort_order: Annotated[
-                SortOrderOrNone,
-                QueryParameter(
-                    title="Field to search",
-                    name="sortOrder",
-                ),
-            ] = config.get("sort_order", "desc"),
-        ) -> OrderBy:
-            return OrderBy(field_name=field_name, sort_order=sort_order)  # type: ignore[arg-type]
-
-        filters[dep_defaults.ORDER_BY_FILTER_DEPENDENCY_KEY] = Provide(provide_order_by, sync_to_thread=False)
+        filters[dep_defaults.ORDER_BY_FILTER_DEPENDENCY_KEY] = Provide(
+            _create_order_by_filter_provider(sort_field, config.get("sort_order", "desc")),
+            sync_to_thread=False,
+        )
 
     # Add not_in filter providers
     if not_in_fields := config.get("not_in_fields"):

advanced_alchemy/filters.py

@@ -63,6 +63,12 @@
 if TYPE_CHECKING:
     from sqlalchemy.orm import InstrumentedAttribute
 
+    FilterFieldName: TypeAlias = Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]
+    InstrumentedField: TypeAlias = Union[ColumnElement[Any], InstrumentedAttribute[Any]]
+else:
+    FilterFieldName: TypeAlias = Any
+    InstrumentedField: TypeAlias = Any
+
 __all__ = (
     "BeforeAfter",
     "CollectionFilter",
@@ -160,9 +166,7 @@ def append_to_statement(
         return statement
 
     @staticmethod
-    def _get_instrumented_attr(
-        model: Any, key: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]"
-    ) -> "Union[ColumnElement[Any], InstrumentedAttribute[Any]]":
+    def _get_instrumented_attr(model: Any, key: FilterFieldName) -> InstrumentedField:
         """Get SQLAlchemy instrumented attribute from model.
 
         Args:
@@ -193,7 +197,7 @@ class BeforeAfter(StatementFilter):
 
     """
 
-    field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]"
+    field_name: FilterFieldName
     """Field name, model attribute, or func expression."""
     before: Optional[datetime.datetime]
     """Filter results where field is earlier than this value."""
@@ -239,7 +243,7 @@ class OnBeforeAfter(StatementFilter):
 
     """
 
-    field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]"
+    field_name: FilterFieldName
     """Field name, model attribute, or func expression."""
     on_or_before: Optional[datetime.datetime]
     """Filter results where field is on or earlier than this value."""
@@ -287,7 +291,7 @@ class CollectionFilter(InAnyFilter, Generic[T]):
     Use ``prefer_any=True`` in ``append_to_statement`` to use the ``ANY`` operator.
     """
 
-    field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]"
+    field_name: FilterFieldName
     """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.
@@ -346,7 +350,7 @@ class NotInCollectionFilter(InAnyFilter, Generic[T]):
 
     """
 
-    field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]"
+    field_name: FilterFieldName
     """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."""
@@ -419,7 +423,7 @@ class NullFilter(StatementFilter):
         - :meth:`sqlalchemy.sql.expression.ColumnOperators.is_`: IS NULL operator
     """
 
-    field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]"
+    field_name: FilterFieldName
     """Field name, model attribute, or func expression."""
 
     def append_to_statement(self, statement: StatementTypeT, model: type[ModelT]) -> StatementTypeT:
@@ -472,7 +476,7 @@ class NotNullFilter(StatementFilter):
         - :meth:`sqlalchemy.sql.expression.ColumnOperators.is_not`: IS NOT NULL operator
     """
 
-    field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]"
+    field_name: FilterFieldName
     """Field name, model attribute, or func expression."""
 
     def append_to_statement(self, statement: StatementTypeT, model: type[ModelT]) -> StatementTypeT:
@@ -556,7 +560,7 @@ class OrderBy(StatementFilter):
         - :meth:`sqlalchemy.sql.expression.ColumnElement.desc`: Descending order
     """
 
-    field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]"
+    field_name: FilterFieldName
     """Field name, model attribute, or func expression (e.g., ``func.random()``)."""
     sort_order: Literal["asc", "desc"] = "asc"
     """Sort direction ("asc" or "desc")."""
@@ -728,7 +732,7 @@ class ComparisonFilter(StatementFilter):
         ValueError: If an invalid operator is provided
     """
 
-    field_name: "Union[str, ColumnElement[Any], InstrumentedAttribute[Any]]"
+    field_name: FilterFieldName
     """Field name, model attribute, or func expression."""
     operator: str
     """Comparison operator to use (one of 'eq', 'ne', 'gt', 'ge', 'lt', 'le')."""

advanced_alchemy/utils/dependencies.py

@@ -12,11 +12,12 @@
 
 from advanced_alchemy.utils.singleton import SingletonMeta
 
-__all__ = ("DependencyCache", "FieldNameType", "FilterConfig", "make_hashable")
+__all__ = ("DependencyCache", "FieldNameType", "FilterConfig", "make_hashable", "normalize_sort_field")
 
 HashableValue = Union[str, int, float, bool, None]
 HashableType = Union[HashableValue, tuple[Any, ...], tuple[tuple[str, Any], ...], tuple[HashableValue, ...]]
 SortOrder = Literal["asc", "desc"]
+SortField = Union[str, set[str], list[str]]
 
 
 class FieldNameType(NamedTuple):
@@ -39,7 +40,7 @@ class FilterConfig(TypedDict):
     """Enable an id filter using the supplied type."""
     id_field: NotRequired[str]
     """Model field storing the primary key or identifier."""
-    sort_field: NotRequired[Union[str, set[str], list[str]]]
+    sort_field: NotRequired[SortField]
     """Default field or fields to use for sorting."""
     sort_order: NotRequired[SortOrder]
     """Default sort order."""
@@ -121,7 +122,10 @@ def make_hashable(value: Any) -> HashableType:
         mapping = cast("Mapping[Any, Any]", value)  # type: ignore[redundant-cast]
         items = [(str(key), make_hashable(mapping[key])) for key in sorted(mapping.keys(), key=str)]
         return tuple(items)
-    if isinstance(value, (list, set)):
+    if isinstance(value, list):
+        values = cast("Iterable[Any]", value)
+        return tuple(make_hashable(item) for item in values)
+    if isinstance(value, set):
         values = cast("Iterable[Any]", value)
         hashable_items = [make_hashable(item) for item in values]
         filtered_items = [item for item in hashable_items if item is not None]
@@ -132,3 +136,12 @@ def make_hashable(value: Any) -> HashableType:
     if isinstance(value, (str, int, float, bool, type(None))):
         return value
     return str(value)
+
+
+def normalize_sort_field(sort_field: SortField) -> str:
+    """Return a scalar default field for a configured sort field value."""
+    if isinstance(sort_field, str):
+        return sort_field
+    if isinstance(sort_field, list):
+        return sort_field[0]
+    return sorted(sort_field)[0]

tests/unit/test_extensions/test_fastapi/test_providers.py

@@ -530,6 +530,30 @@ def test_provide_filters_normalizes_list_field_config() -> None:
     }
 
 
+def test_provide_filters_uses_scalar_default_from_list_sort_field() -> None:
+    """Test list-based sort fields produce a scalar default order field."""
+    deps = provide_filters({"sort_field": ["name", "id"]})
+
+    app = FastAPI()
+
+    @app.get("/items")
+    async def get_items(filters: Annotated[list[FilterTypes], Depends(deps)]) -> list[dict[str, str]]:
+        return [
+            {"field_name": cast(str, filter_.field_name), "sort_order": filter_.sort_order}
+            for filter_ in filters
+            if isinstance(filter_, OrderBy)
+        ]
+
+    client = TestClient(app)
+    response = client.get("/items")
+    overridden_response = client.get("/items?orderBy=id&sortOrder=asc")
+
+    assert response.status_code == 200
+    assert response.json() == [{"field_name": "name", "sort_order": "desc"}]
+    assert overridden_response.status_code == 200
+    assert overridden_response.json() == [{"field_name": "id", "sort_order": "asc"}]
+
+
 def test_openapi_schema_edge_cases() -> None:
     """Test OpenAPI schema generation for edge cases and special configurations."""
     # Test with minimal configuration

tests/unit/test_extensions/test_litestar/test_providers.py

@@ -438,6 +438,83 @@ def test_order_by_filter() -> None:
     assert f.sort_order == "desc"
 
 
+def test_order_by_filter_uses_scalar_default_from_list_sort_field() -> None:
+    """Test list-based sort fields produce a scalar default order field."""
+    deps = _create_statement_filters({"sort_field": ["name", "id"]})
+
+    provider_func = deps["order_by_filter"].dependency
+    f = provider_func()
+
+    assert isinstance(f, OrderBy)
+    assert f.field_name == "name"
+    f.append_to_statement(select(DITestModel), DITestModel)
+
+
+def test_order_by_filter_uses_scalar_default_from_set_sort_field() -> None:
+    """Test set-based sort fields produce a deterministic scalar default order field."""
+    deps = _create_statement_filters({"sort_field": {"name", "id"}})
+
+    provider_func = deps["order_by_filter"].dependency
+    f = provider_func()
+
+    assert isinstance(f, OrderBy)
+    assert f.field_name == "id"
+    f.append_to_statement(select(DITestModel), DITestModel)
+
+
+def test_individual_order_by_dependency_injection_skips_runtime_field_validation() -> None:
+    """Test individual Litestar filter dependencies can be injected directly."""
+
+    @get("/", dependencies=create_filter_dependencies({"sort_field": "name"}))
+    async def handler(order_by_filter: OrderBy) -> dict[str, str]:
+        return {"field_name": cast(str, order_by_filter.field_name), "sort_order": order_by_filter.sort_order}
+
+    with TestClient(Litestar([handler])) as client:
+        response = client.get("/")
+        overridden_response = client.get("/?orderBy=id&sortOrder=asc")
+
+    assert response.status_code == 200
+    assert response.json() == {"field_name": "name", "sort_order": "desc"}
+    assert overridden_response.status_code == 200
+    assert overridden_response.json() == {"field_name": "id", "sort_order": "asc"}
+
+
+def test_litestar_openapi_schema_uses_typed_sort_parameters() -> None:
+    """Test sort parameters keep concrete OpenAPI schemas for collection defaults."""
+    filter_dependencies = create_filter_dependencies({"sort_field": ["name", "id"], "sort_order": "asc"})
+
+    @get("/test")
+    async def handler(filters: Annotated[list[FilterTypes], Dependency(skip_validation=True)]) -> list[str]:
+        return [type(filter_).__name__ for filter_ in filters]
+
+    app = Litestar(
+        route_handlers=[handler],
+        signature_namespace=signature_namespace_values,
+        dependencies=filter_dependencies,
+        openapi_config=OpenAPIConfig(title="Test API", version="1.0.0", path="/schema"),
+    )
+
+    with TestClient(app) as client:
+        response = client.get("/schema/openapi.json")
+
+    assert response.status_code == 200
+    parameters = response.json()["paths"]["/test"]["get"]["parameters"]
+    order_by_param = next(param for param in parameters if param["name"] == "orderBy")
+    sort_order_param = next(param for param in parameters if param["name"] == "sortOrder")
+
+    assert order_by_param["schema"] == {
+        "oneOf": [{"type": "string", "default": "name"}, {"type": "null"}],
+        "title": "Order by field",
+        "default": "name",
+    }
+    assert sort_order_param["schema"] == {
+        "type": ["null", "string"],
+        "enum": ["asc", "desc", None],
+        "title": "Field to search",
+        "default": "asc",
+    }
+
+
 def test_not_in_filter() -> None:
     """Test creating not_in filter dependency."""
     deps = _create_statement_filters({"not_in_fields": ["status"]})