Adapter paginator and pagination operations to both controllers route functions and functional routes

Author: eadwinCode

ninja_extra/ordering.py

@@ -18,7 +18,8 @@
 
 from asgiref.sync import sync_to_async
 from django.db.models import QuerySet
-from ninja import Field, Query, Schema
+from django.http import HttpRequest
+from ninja import Field, P, Query, Schema
 from ninja.constants import NOT_SET
 from ninja.signature import is_async
 from pydantic import BaseModel
@@ -65,14 +66,24 @@ def __init__(
     ) -> None:
         super().__init__(pass_parameter=pass_parameter)
         self.ordering_fields = ordering_fields or "__all__"
+        self.Input = self.create_input(ordering_fields)  # type:ignore
+
+    def create_input(self, ordering_fields: Optional[List[str]]) -> Type[Input]:
+        if ordering_fields:
+
+            class DynamicInput(Ordering.Input):
+                ordering: Query[Optional[str], P(default=",".join(ordering_fields))]  # type:ignore[type-arg,valid-type]
+
+            return DynamicInput
+        return Ordering.Input
 
     def ordering_queryset(
         self, items: Union[QuerySet, List], ordering_input: Input
     ) -> Union[QuerySet, List]:
-        ordering = self.get_ordering(items, ordering_input.ordering)
-        if ordering:
+        ordering_ = self.get_ordering(items, ordering_input.ordering)
+        if ordering_:
             if isinstance(items, QuerySet):  # type:ignore
-                return items.order_by(*ordering)
+                return items.order_by(*ordering_)
             elif isinstance(items, list) and items:
 
                 def multisort(xs: List, specs: List[Tuple[str, bool]]) -> List:
@@ -85,7 +96,7 @@ def multisort(xs: List, specs: List[Tuple[str, bool]]) -> List:
                     items,
                     [
                         (o[int(o.startswith("-")) :], o.startswith("-"))
-                        for o in ordering
+                        for o in ordering_
                     ],
                 )
         return items
@@ -201,49 +212,51 @@ def __init__(
         self.view_func = view_func
 
         orderator_view = self.get_view_function()
-        _ninja_contribute_args: List[Tuple] = getattr(
-            self.view_func, "_ninja_contribute_args", []
-        )
-        orderator_view._ninja_contribute_args = (  # type:ignore[attr-defined]
-            _ninja_contribute_args
-        )
+        self.as_view = wraps(view_func)(orderator_view)
         add_ninja_contribute_args(
-            orderator_view,
+            self.as_view,
             (
                 self.orderator_kwargs_name,
                 self.orderator.Input,
                 self.orderator.InputSource,
             ),
         )
         orderator_view.orderator_operation = self  # type:ignore[attr-defined]
-        self.as_view = wraps(view_func)(orderator_view)
 
     @property
     def view_func_has_kwargs(self) -> bool:  # pragma: no cover
         return self.orderator.pass_parameter is not None
 
     def get_view_function(self) -> Callable:
-        def as_view(controller: "ControllerBase", *args: Any, **kw: Any) -> Any:
+        def as_view(
+            request_or_controller: Union["ControllerBase", HttpRequest],
+            *args: Any,
+            **kw: Any,
+        ) -> Any:
             func_kwargs = dict(**kw)
             ordering_params = func_kwargs.pop(self.orderator_kwargs_name)
             if self.orderator.pass_parameter:
                 func_kwargs[self.orderator.pass_parameter] = ordering_params
 
-            items = self.view_func(controller, *args, **func_kwargs)
+            items = self.view_func(request_or_controller, *args, **func_kwargs)
             return self.orderator.ordering_queryset(items, ordering_params)
 
         return as_view
 
 
 class AsyncOrderatorOperation(OrderatorOperation):
     def get_view_function(self) -> Callable:
-        async def as_view(controller: "ControllerBase", *args: Any, **kw: Any) -> Any:
+        async def as_view(
+            request_or_controller: Union["ControllerBase", HttpRequest],
+            *args: Any,
+            **kw: Any,
+        ) -> Any:
             func_kwargs = dict(**kw)
             ordering_params = func_kwargs.pop(self.orderator_kwargs_name)
             if self.orderator.pass_parameter:
                 func_kwargs[self.orderator.pass_parameter] = ordering_params
 
-            items = await self.view_func(controller, *args, **func_kwargs)
+            items = await self.view_func(request_or_controller, *args, **func_kwargs)
             ordering_queryset = cast(
                 Callable, sync_to_async(self.orderator.ordering_queryset)
             )

ninja_extra/pagination/operations.py

@@ -3,14 +3,15 @@
     TYPE_CHECKING,
     Any,
     Callable,
-    List,
-    Tuple,
+    Union,
     cast,
 )
 
 from asgiref.sync import sync_to_async
+from django.http import HttpRequest
 from ninja.pagination import PaginationBase
 
+from ninja_extra.controllers.route.context import RouteContext
 from ninja_extra.shortcuts import add_ninja_contribute_args
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -30,60 +31,69 @@ def __init__(
         self.view_func = view_func
 
         paginator_view = self.get_view_function()
-        _ninja_contribute_args: List[Tuple] = getattr(
-            self.view_func, "_ninja_contribute_args", []
-        )
-        paginator_view._ninja_contribute_args = (  # type:ignore[attr-defined]
-            _ninja_contribute_args
-        )
+        self.as_view = wraps(view_func)(paginator_view)
         add_ninja_contribute_args(
-            paginator_view,
+            self.as_view,
             (
                 self.paginator_kwargs_name,
                 self.paginator.Input,
                 self.paginator.InputSource,
             ),
         )
         paginator_view.paginator_operation = self  # type:ignore[attr-defined]
-        self.as_view = wraps(view_func)(paginator_view)
 
     @property
     def view_func_has_kwargs(self) -> bool:  # pragma: no cover
         return self.paginator.pass_parameter is not None
 
     def get_view_function(self) -> Callable:
-        def as_view(controller: "ControllerBase", *args: Any, **kw: Any) -> Any:
+        def as_view(
+            request_or_controller: Union["ControllerBase", HttpRequest],
+            *args: Any,
+            **kw: Any,
+        ) -> Any:
             func_kwargs = dict(**kw)
             pagination_params = func_kwargs.pop(self.paginator_kwargs_name)
             if self.paginator.pass_parameter:
                 func_kwargs[self.paginator.pass_parameter] = pagination_params
 
-            items = self.view_func(controller, *args, **func_kwargs)
-            assert (
-                controller.context and controller.context.request
-            ), "Request object is None"
+            items = self.view_func(request_or_controller, *args, **func_kwargs)
+            if hasattr(request_or_controller, "context") and isinstance(
+                request_or_controller.context, RouteContext
+            ):
+                request = request_or_controller.context.request
+                assert request, "Request object is None"
+            else:
+                request = request_or_controller
             params = dict(kw)
-            params["request"] = controller.context.request
+            params["request"] = request
             return self.paginator.paginate_queryset(items, **params)
 
         return as_view
 
 
 class AsyncPaginatorOperation(PaginatorOperation):
     def get_view_function(self) -> Callable:
-        async def as_view(controller: "ControllerBase", *args: Any, **kw: Any) -> Any:
+        async def as_view(
+            request_or_controller: Union["ControllerBase", HttpRequest],
+            *args: Any,
+            **kw: Any,
+        ) -> Any:
             func_kwargs = dict(**kw)
             pagination_params = func_kwargs.pop(self.paginator_kwargs_name)
             if self.paginator.pass_parameter:
                 func_kwargs[self.paginator.pass_parameter] = pagination_params
 
-            items = await self.view_func(controller, *args, **func_kwargs)
-            assert (
-                controller.context and controller.context.request
-            ), "Request object is None"
-
+            items = await self.view_func(request_or_controller, *args, **func_kwargs)
+            if hasattr(request_or_controller, "context") and isinstance(
+                request_or_controller.context, RouteContext
+            ):
+                request = request_or_controller.context.request
+                assert request, "Request object is None"
+            else:
+                request = request_or_controller
             params = dict(kw)
-            params["request"] = controller.context.request
+            params["request"] = request
             paginate_queryset = cast(
                 Callable, sync_to_async(self.paginator.paginate_queryset)
             )

ninja_extra/searching.py

@@ -91,8 +91,8 @@ def __init__(
         search_fields: Optional[List[str]] = None,
         pass_parameter: Optional[str] = None,
     ) -> None:
+        super().__init__(pass_parameter=pass_parameter)
         self.search_fields = search_fields or []
-        self.pass_parameter = pass_parameter
 
     def searching_queryset(
         self, items: Union[QuerySet, List], searching_input: Input
@@ -240,22 +240,16 @@ def __init__(
         self.view_func = view_func
 
         searcherator_view = self.get_view_function()
-        _ninja_contribute_args: List[Tuple] = getattr(
-            self.view_func, "_ninja_contribute_args", []
-        )
-        searcherator_view._ninja_contribute_args = (  # type:ignore[attr-defined]
-            _ninja_contribute_args
-        )
+        self.as_view = wraps(view_func)(searcherator_view)
         add_ninja_contribute_args(
-            searcherator_view,
+            self.as_view,
             (
                 self.searcherator_kwargs_name,
                 self.searcherator.Input,
                 self.searcherator.InputSource,
             ),
         )
         searcherator_view.searcherator_operation = self  # type:ignore[attr-defined]
-        self.as_view = wraps(view_func)(searcherator_view)
 
     @property
     def view_func_has_kwargs(self) -> bool:  # pragma: no cover

tests/test_ordering.py

@@ -189,16 +189,17 @@ def test_case4(self):
         assert response[0]["title"] == "title_2"
 
         schema = api.get_openapi_schema()["paths"]["/api/items_4"]["get"]
-        # print(schema)
+        print(schema["parameters"])
         assert schema["parameters"] == [
             {
                 "in": "query",
                 "name": "ordering",
-                "required": False,
                 "schema": {
                     "anyOf": [{"type": "string"}, {"type": "null"}],
+                    "default": "title",
                     "title": "Ordering",
                 },
+                "required": False,
             }
         ]
 
@@ -382,11 +383,12 @@ async def test_case4(self):
                 {
                     "in": "query",
                     "name": "ordering",
-                    "required": False,
                     "schema": {
                         "anyOf": [{"type": "string"}, {"type": "null"}],
+                        "default": "title",
                         "title": "Ordering",
                     },
+                    "required": False,
                 }
             ]
 

tests/test_pagination.py

@@ -3,7 +3,7 @@
 
 import django
 import pytest
-from ninja import Schema
+from ninja import NinjaAPI, Schema
 
 from ninja_extra import NinjaExtraAPI, api_controller, route
 from ninja_extra.controllers import RouteFunction
@@ -385,5 +385,52 @@ async def test_case5(self):
             assert data["items"] == ITEMS[10:20]
 
 
-def test_pagination_extra_get_schema():
-    pass
+def test_pagination_extra_with_ninja_api():
+    app = NinjaAPI()
+
+    @app.get("/items_2", response=NinjaPaginationResponseSchema[int])
+    @paginate()  # with brackets (should use default pagination)
+    def items_2(request, someparam: int = 0):
+        # also having custom param `someparam` - that should not be lost
+        return FakeQuerySet()
+
+    @app.get("/items_3")
+    @paginate(CustomPagination, pass_parameter="pass_kwargs")
+    def items_3(request, **kwargs):
+        return ITEMS
+
+    _client = TestClient(app)
+
+    response = _client.get("/items_3?skip=5")
+    assert response.json() == ITEMS[5:10]
+
+    response = _client.get("/items_2?limit=10").json()
+    assert response.get("items")
+    assert response["items"] == ITEMS[:10]
+
+
+@pytest.mark.skipif(django.VERSION < (3, 1), reason="requires django 3.1 or higher")
+@pytest.mark.asyncio
+async def test_pagination_extra_with_ninja_api_async():
+    app = NinjaAPI()
+
+    @app.get("/items_2", response=NinjaPaginationResponseSchema[int])
+    @paginate()  # with brackets (should use default pagination)
+    async def items_2(request, someparam: int = 0):
+        # also having custom param `someparam` - that should not be lost
+        return FakeQuerySet()
+
+    @app.get("/items_3")
+    @paginate(CustomPagination, pass_parameter="pass_kwargs")
+    async def items_3(request, **kwargs):
+        return ITEMS
+
+    _client = TestAsyncClient(app)
+
+    response = await _client.get("/items_3?skip=5")
+    assert response.json() == ITEMS[5:10]
+
+    response = await _client.get("/items_2?limit=10")
+    result = response.json()
+    assert result.get("items")
+    assert result["items"] == ITEMS[:10]