add searching and ordering

QtiQla · QtiQla · commit 17c67c8a9e7d · 2023-02-28T18:09:53.000+05:00
diff --git a/ninja_extra/conf/settings.py b/ninja_extra/conf/settings.py
@@ -21,6 +21,8 @@ def __init__(self, data: dict) -> None:
         "ninja_extra.throttling.UserRateThrottle",
     ],
     THROTTLE_RATES={"user": None, "anon": None},
+    ORDERING_CLASS="ninja_extra.ordering.Ordering",
+    SEARCHING_CLASS="ninja_extra.searching.Search"
 )
 
 USER_SETTINGS = UserDefinedSettingsMapper(
@@ -43,6 +45,12 @@ class Config:
     THROTTLE_CLASSES: List[Any] = []
     NUM_PROXIES: Optional[int] = None
     INJECTOR_MODULES: List[Any] = []
+    ORDERING_CLASS: Any = Field(
+        "ninja_extra.ordering.Ordering",
+    )
+    SEARCHING_CLASS: Any = Field(
+        "ninja_extra.searching.Search",
+    )
 
     @validator("INJECTOR_MODULES", pre=True)
     def pre_injector_module_validate(cls, value: Any) -> Any:
diff --git a/ninja_extra/ordering.py b/ninja_extra/ordering.py
@@ -0,0 +1,106 @@
+from functools import wraps
+from inspect import isfunction as is_func
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    List,
+    Optional,
+    Tuple,
+    Type,
+    Union,
+)
+
+from django.db.models import QuerySet
+from ninja import Field, Query, Schema
+from ninja.constants import NOT_SET
+
+from ninja_extra.conf import settings
+
+if TYPE_CHECKING:  # pragma: no cover
+    from ninja_extra.controllers import ControllerBase
+
+
+class Ordering:
+
+    class Input(Schema):
+        ordering: Optional[str] = Field(None)
+
+    input_source = Query(...)
+
+    def __init__(self, ordering_fields: Optional[Union[str, List[str]]] = None, pass_parameter: Optional[str] = None) -> None:
+        self.ordering_fields = ordering_fields or '__all__'
+        self.pass_parameter = pass_parameter
+
+    def ordering_queryset(self, queryset: QuerySet, ordering: Input):
+        ordering = self.get_ordering(queryset, ordering.ordering)
+        if ordering:
+            return queryset.order_by(*ordering)
+        return queryset
+
+    def get_ordering(self, queryset: QuerySet, value: Optional[str]) -> Optional[List[str]]:
+        if value:
+            fields = [param.strip() for param in value.split(',')]
+            return self.remove_invalid_fields(queryset, fields)
+        return []
+
+    def remove_invalid_fields(self, queryset, fields):
+        valid_fields = [item[0] for item in self.get_valid_fields(queryset)]
+
+        def term_valid(term):
+            if term.startswith("-"):
+                term = term[1:]
+            return term in valid_fields
+
+        return [term for term in fields if term_valid(term)]
+
+    def get_valid_fields(self, queryset):
+        valid_fields = self.ordering_fields
+        if valid_fields == '__all__':
+            valid_fields = [(field.name, field.verbose_name) for field in queryset.model._meta.fields]
+            valid_fields += [(key, key.title().split('__')) for key in queryset.query.annotations]
+        else:
+            valid_fields = [(item, item) if isinstance(item, str) else item for item in valid_fields]
+        return valid_fields
+
+
+def ordering(func_or_ordering_class: Type[Ordering] = NOT_SET, **ordering_params: Any) -> Callable:
+    isfunction = is_func(func_or_ordering_class)
+    isnotset = func_or_ordering_class == NOT_SET
+    ordering_class: Type[Ordering] = settings.ORDERING_CLASS
+    if isfunction:
+        return _inject_orderator(func_or_ordering_class, ordering_class)
+    if not isnotset:
+        ordering_class = func_or_ordering_class
+
+    def wrapper(func: Callable[..., Any]) -> Any:
+        return _inject_orderator(func, ordering_class, **ordering_params)
+
+    return wrapper
+
+
+def _inject_orderator(
+    func: Callable[..., Any],
+    ordering_class: Type[Ordering],
+    **ordering_params: Any,
+) -> Callable[..., Any]:
+    orderator: Ordering = ordering_class(**ordering_params)
+    orderator_kwargs_name = "ordering"
+    _ninja_contribute_args: List[Tuple] = getattr(func, "_ninja_contribute_args", [])
+
+    @wraps(func)
+    def view_with_ordering(controller: "ControllerBase", *args: Any, **kw: Any) -> Any:
+        func_kwargs = dict(**kw)
+        ordering = func_kwargs.pop(orderator_kwargs_name)
+        if orderator.pass_parameter:
+            func_kwargs[orderator.pass_parameter] = ordering_params
+        items = func(controller, *args, **func_kwargs)
+        return orderator.ordering_queryset(items, ordering)
+
+    _ninja_contribute_args.append((
+        orderator_kwargs_name,
+        orderator.Input,
+        orderator.input_source,
+    ))
+    view_with_ordering._ninja_contribute_args = _ninja_contribute_args
+    return view_with_ordering
diff --git a/ninja_extra/pagination.py b/ninja_extra/pagination.py
@@ -2,7 +2,7 @@
 import logging
 from collections import OrderedDict
 from functools import wraps
-from typing import TYPE_CHECKING, Any, Callable, Optional, Type, Union, cast, overload
+from typing import TYPE_CHECKING, Any, Callable, Optional, Type, Union, cast, overload, List, Tuple
 
 from asgiref.sync import sync_to_async
 from django.core.paginator import InvalidPage, Page, Paginator
@@ -180,6 +180,8 @@ 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 = _ninja_contribute_args
         add_ninja_contribute_args(
             paginator_view,
             (
diff --git a/ninja_extra/searching.py b/ninja_extra/searching.py
@@ -0,0 +1,104 @@
+import inspect
+import operator
+from functools import reduce, wraps
+from typing import TYPE_CHECKING, Any, Callable, List, Optional, Tuple, Type
+
+from django.db.models import Q, QuerySet
+from django.db.models.constants import LOOKUP_SEP
+from ninja import Field, Query, Schema
+from ninja.constants import NOT_SET
+
+from ninja_extra.conf import settings
+
+if TYPE_CHECKING:  # pragma: no cover
+    from ninja_extra.controllers import ControllerBase
+
+
+class Search:
+
+    class Input(Schema):
+        search: Optional[str] = Field(None)
+
+    lookup_prefixes = {
+        '^': 'istartswith',
+        '=': 'iexact',
+        '@': 'search',
+        '$': 'iregex',
+    }
+    input_source = Query(...)
+
+    def __init__(self, search_fields: List[str], pass_parameter: Optional[str] = None) -> None:
+        self.search_fields = search_fields
+        self.pass_parameter = pass_parameter
+
+    def search_queryset(self, queryset: QuerySet, searching: Input):
+        if searching.search:
+            search_terms = self.get_search_terms(searching.search)
+
+            if not self.search_fields or not search_terms:
+                return queryset
+
+            orm_lookups = [self.construct_search(str(search_field)) for search_field in self.search_fields]
+            conditions = []
+            for search_term in search_terms:
+                queries = [Q(**{orm_lookup: search_term}) for orm_lookup in orm_lookups]
+                conditions.append(reduce(operator.or_, queries))
+            queryset = queryset.filter(reduce(operator.and_, conditions))
+        return queryset
+
+    def get_search_terms(self, value: str):
+        if value:
+            value = value.replace('\x00', '')  # strip null characters
+            value = value.replace(',', ' ')
+            return value.split()
+        return ''
+
+    def construct_search(self, field_name):
+        lookup = self.lookup_prefixes.get(field_name[0])
+        if lookup:
+            field_name = field_name[1:]
+        else:
+            lookup = 'icontains'
+        return LOOKUP_SEP.join([field_name, lookup])
+
+
+def searching(func_or_searching_class: Type[Search] = NOT_SET, **searching_params: Any) -> Callable:
+    isfunction = inspect.isfunction(func_or_searching_class)
+    isnotset = func_or_searching_class == NOT_SET
+    searching_class: Type[Search] = settings.SEARCHING_CLASS
+    if isfunction:
+        return _inject_searcherator(func_or_searching_class, searching_class)
+    if not isnotset:
+        searching_class = func_or_searching_class
+
+    def wrapper(func: Callable[..., Any]) -> Any:
+        return _inject_searcherator(func, searching_class, **searching_params)
+
+    return wrapper
+
+
+def _inject_searcherator(
+    func: Callable[..., Any],
+    searching_class: Type[Search],
+    **searching_params: Any,
+) -> Callable[..., Any]:
+    searcherator: Search = searching_class(**searching_params)
+    searcherator_kwargs_name = "searching"
+    _ninja_contribute_args: List[Tuple] = getattr(func, "_ninja_contribute_args", [])
+
+    @wraps(func)
+    def view_with_searching(controller: "ControllerBase", *args: Any, **kw: Any) -> Any:
+        func_kwargs = dict(**kw)
+        searching = func_kwargs.pop(searcherator_kwargs_name)
+        if searcherator.pass_parameter:
+            func_kwargs[searcherator.pass_parameter] = searching_params
+        items = func(controller, *args, **func_kwargs)
+        return searcherator.search_queryset(items, searching)
+
+    _ninja_contribute_args.append((
+        searcherator_kwargs_name,
+        searcherator.Input,
+        searcherator.input_source,
+    ))
+    view_with_searching._ninja_contribute_args = _ninja_contribute_args
+    return view_with_searching
