new ordering class

Author: pcrespov

packages/models-library/src/models_library/rest_ordering.py

@@ -1,32 +1,41 @@
-from enum import Enum
-from typing import Annotated
+from typing import Annotated, Generic
 
+from common_library.basic_types import DEFAULT_FACTORY
 from common_library.json_serialization import json_dumps
-from pydantic import BaseModel, BeforeValidator, ConfigDict, Field, field_validator
+from pydantic import (
+    BaseModel,
+    BeforeValidator,
+    ConfigDict,
+    Field,
+    field_validator,
+)
 
 from .basic_types import IDStr
+from .list_operations import OrderClause, OrderDirection, TField, check_ordering_list
 from .rest_base import RequestParameters
-from .utils.common_validators import parse_json_pre_validator
+from .utils.common_validators import (
+    parse_json_pre_validator,
+)
 
-
-class OrderDirection(str, Enum):
-    ASC = "asc"
-    DESC = "desc"
+__all__: tuple[str, ...] = ("OrderDirection",)
 
 
 class OrderBy(BaseModel):
-    # Based on https://google.aip.dev/132#ordering
-    field: IDStr = Field(..., description="field name identifier")
-    direction: OrderDirection = Field(
-        default=OrderDirection.ASC,
-        description=(
-            f"As [A,B,C,...] if `{OrderDirection.ASC.value}`"
-            f" or [Z,Y,X, ...] if `{OrderDirection.DESC.value}`"
+    # NOTE: use instead OrderClause[TField] where TField is Literal of valid fields
+    field: Annotated[IDStr, Field(description="field name identifier")]
+    direction: Annotated[
+        OrderDirection,
+        Field(
+            description=(
+                f"As [A,B,C,...] if `{OrderDirection.ASC.value}`"
+                f" or [Z,Y,X, ...] if `{OrderDirection.DESC.value}`"
+            )
         ),
-    )
+    ] = OrderDirection.ASC
 
 
 class _BaseOrderQueryParams(RequestParameters):
+    # Use OrderingQueryParams instead for more flexible ordering
     order_by: OrderBy
 
 
@@ -91,30 +100,93 @@ def _check_ordering_field_and_map(cls, v):
             return _ordering_fields_api_to_column_map.get(v) or v
 
     assert "json_schema_extra" in _OrderBy.model_config  # nosec
-    assert isinstance(_OrderBy.model_config["json_schema_extra"], dict)  # nosec
-    assert isinstance(  # nosec
-        _OrderBy.model_config["json_schema_extra"]["examples"], list
-    )
-    order_by_example = _OrderBy.model_config["json_schema_extra"]["examples"][0]
+
+    order_by_example = _OrderBy.model_json_schema()["examples"][0]
     order_by_example_json = json_dumps(order_by_example)
+
     assert _OrderBy.model_validate(order_by_example), "Example is invalid"  # nosec
 
     converted_default = _OrderBy.model_validate(
         # NOTE: enforces ordering_fields_api_to_column_map
         default.model_dump()
     )
 
-    class _OrderQueryParams(_BaseOrderQueryParams):
-        order_by: Annotated[_OrderBy, BeforeValidator(parse_json_pre_validator)] = (
+    class _OrderJsonQueryParams(_BaseOrderQueryParams):
+        order_by: Annotated[
+            _OrderBy,
+            BeforeValidator(parse_json_pre_validator),
             Field(
-                default=converted_default,
                 description=(
                     f"Order by field (`{msg_field_options}`) and direction (`{msg_direction_options}`). "
                     f"The default sorting order is `{json_dumps(default)}`."
                 ),
                 examples=[order_by_example],
                 json_schema_extra={"example_json": order_by_example_json},
-            )
-        )
+            ),
+        ] = converted_default
+
+    return _OrderJsonQueryParams
+
+
+def _parse_order_by(v):
+    if not v:
+        return []
+
+    if isinstance(v, list):
+        v = ",".join(v)
+
+    if not isinstance(v, str):
+        msg = "order_by must be a string"
+        raise TypeError(msg)
 
-    return _OrderQueryParams
+    # 1. from comma-separated string to list of OrderClause
+    clauses = []
+    for t in v.split(","):
+        token = t.strip()
+        if not token:
+            continue
+        if token.startswith("-"):
+            clauses.append((token[1:], OrderDirection.DESC))
+        elif token.startswith("+"):
+            clauses.append((token[1:], OrderDirection.ASC))
+        else:
+            clauses.append((token, OrderDirection.ASC))
+
+    # 2. check for duplicates and conflicting directions
+    return [
+        {"field": field, "direction": direction}
+        for field, direction in check_ordering_list(clauses)
+    ]
+
+
+class OrderingQueryParams(BaseModel, Generic[TField]):
+    """
+    This class is designed to parse query parameters for ordering results in an API request.
+
+    It supports multiple ordering clauses and allows for flexible sorting options.
+
+    NOTE: It only parses strings and validates into list[OrderClause[TField]]
+    where TField is a type variable representing valid field names.
+
+
+    For example:
+
+        /my/path?order_by=field1,-field2,+field3
+
+    would sort by field1 ascending, field2 descending, and field3 ascending.
+    """
+
+    order_by: Annotated[
+        list[OrderClause[TField]],
+        BeforeValidator(_parse_order_by),
+        Field(default_factory=list),
+    ] = DEFAULT_FACTORY
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "examples": [
+                {"order_by": "-created_at,name,+gender"},
+                {"order_by": ""},
+            ],
+        }
+    )

packages/models-library/tests/test_rest_ordering.py

@@ -1,11 +1,14 @@
 import pickle
+from typing import Literal
 
 import pytest
 from common_library.json_serialization import json_dumps
 from models_library.basic_types import IDStr
 from models_library.rest_ordering import (
     OrderBy,
+    OrderClause,
     OrderDirection,
+    OrderingQueryParams,
     create_ordering_query_model_class,
 )
 from pydantic import (
@@ -236,3 +239,47 @@ def test_ordering_query_parse_json_pre_validator():
     assert error["loc"] == ("order_by",)
     assert error["type"] == "value_error"
     assert error["input"] == bad_json_value
+
+
+def test_ordering_query_params_parsing():
+    """Test OrderingQueryParams parsing from URL query format like ?order_by=-created_at,name,+gender"""
+
+    # Define allowed fields using Literal type
+    ValidField = Literal["created_at", "name", "gender"]
+
+    class TestOrderingParams(OrderingQueryParams[ValidField]):
+        pass
+
+    # Test parsing from comma-separated string
+    params = TestOrderingParams.model_validate({"order_by": "-created_at,name,+gender"})
+
+    assert params.order_by == [
+        OrderClause[ValidField](field="created_at", direction=OrderDirection.DESC),
+        OrderClause[ValidField](field="name", direction=OrderDirection.ASC),
+        OrderClause[ValidField](field="gender", direction=OrderDirection.ASC),
+    ]
+
+
+def test_ordering_query_params_validation_error_with_invalid_fields():
+    """Test that OrderingQueryParams raises ValidationError when invalid fields are used"""
+
+    # Define allowed fields using Literal type
+    ValidField = Literal["created_at", "name"]
+
+    class TestOrderingParams(OrderingQueryParams[ValidField]):
+        pass
+
+    # Test with invalid field should raise ValidationError
+    with pytest.raises(ValidationError) as err_info:
+        TestOrderingParams.model_validate(
+            {"order_by": "-created_at,invalid_field,name"}
+        )
+
+    # Verify the validation error details
+    exc = err_info.value
+    assert exc.error_count() == 1
+
+    error = exc.errors()[0]
+    assert error["loc"] == ("order_by", 1, "field")
+    assert error["type"] == "literal_error"
+    assert error["input"] == "invalid_field"