Merge pull request #61 from QtiQla/master

eadwinCode · web-flow · commit a9b2ee2f9bf9 · 2023-04-14T22:06:21.000+01:00
add list support for filtering
diff --git a/docs/tutorial/ordering.md b/docs/tutorial/ordering.md
@@ -1,6 +1,6 @@
 # **Ordering**
 
-**Django Ninja Extra** provides an intuitive ordering model using `ordering` decoration from the Django-Ninja-Extra ordering module. It expects a Queryset from as a route function result.
+**Django Ninja Extra** provides an intuitive ordering model using `ordering` decoration from the Django-Ninja-Extra ordering module. It expects a Queryset or a List from as a route function result.
 
 > This feature was inspired by the [DRF OrderingFilter](https://www.django-rest-framework.org/api-guide/filtering/#orderingfilter)
 
@@ -61,7 +61,7 @@ class UserController:
     @route.get('/all-sort', response=List[UserSchema])
     @ordering
     def get_users_with_all_field_ordering(self):
-        return user_model.objects.all()
+        return [u for u in user_model.objects.all()]
 
 
 api = NinjaExtraAPI(title='Ordering Test')
diff --git a/docs/tutorial/searching.md b/docs/tutorial/searching.md
@@ -1,6 +1,6 @@
 # **Searching**
 
-**Django Ninja Extra** provides an intuitive searching model using `searching` decoration from the Django-Ninja-Extra searching module. It expects a Queryset from as a route function result.
+**Django Ninja Extra** provides an intuitive searching model using `searching` decoration from the Django-Ninja-Extra searching module. It expects a Queryset or a List from as a route function result.
 
 > This feature was inspired by the [DRF SearchFilter](https://www.django-rest-framework.org/api-guide/filtering/#searchfilter)
 
@@ -69,7 +69,7 @@ class UserController:
     @route.get('/iexact-email', response=List[UserSchema])
     @searching(search_fields=['=email'])
     def get_users_with_search_iexact_email(self):
-        return user_model.objects.all()
+        return [u for u in user_model.objects.all()]
 
 
 api = NinjaExtraAPI(title='Searching Test')
diff --git a/ninja_extra/ordering.py b/ninja_extra/ordering.py
@@ -2,6 +2,7 @@
 import logging
 from abc import ABC, abstractmethod
 from functools import wraps
+from operator import attrgetter, itemgetter
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -10,6 +11,7 @@
     Optional,
     Tuple,
     Type,
+    Union,
     cast,
     overload,
 )
@@ -19,6 +21,7 @@
 from ninja import Field, Query, Schema
 from ninja.constants import NOT_SET
 from ninja.signature import is_async
+from pydantic import BaseModel
 
 from ninja_extra.conf import settings
 from ninja_extra.shortcuts import add_ninja_contribute_args
@@ -47,7 +50,9 @@ def __init__(self, *, pass_parameter: Optional[str] = None, **kwargs: Any) -> No
         self.pass_parameter = pass_parameter
 
     @abstractmethod
-    def ordering_queryset(self, queryset: QuerySet, ordering_input: Any) -> QuerySet:
+    def ordering_queryset(
+        self, items: Union[QuerySet, List], ordering_input: Any
+    ) -> Union[QuerySet, List]:
         ...
 
 
@@ -70,20 +75,42 @@ class DynamicInput(Ordering.Input):
 
         return DynamicInput
 
-    def ordering_queryset(self, queryset: QuerySet, ordering_input: Input) -> QuerySet:
-        ordering = self.get_ordering(queryset, ordering_input.ordering)
+    def ordering_queryset(
+        self, items: Union[QuerySet, List], ordering_input: Input
+    ) -> Union[QuerySet, List]:
+        ordering = self.get_ordering(items, ordering_input.ordering)
         if ordering:
-            return queryset.order_by(*ordering)
-        return queryset
-
-    def get_ordering(self, queryset: QuerySet, value: Optional[str]) -> List[str]:
+            if isinstance(items, QuerySet):  # type:ignore
+                return items.order_by(*ordering)
+            elif isinstance(items, list) and items:
+
+                def multisort(xs: List, specs: List[Tuple[str, bool]]) -> List:
+                    orerator = itemgetter if isinstance(xs[0], dict) else attrgetter
+                    for key, reverse in specs:
+                        xs.sort(key=orerator(key), reverse=reverse)
+                    return xs
+
+                return multisort(
+                    items,
+                    [
+                        (o[int(o.startswith("-")) :], o.startswith("-"))
+                        for o in ordering
+                    ],
+                )
+        return items
+
+    def get_ordering(
+        self, items: Union[QuerySet, List], value: Optional[str]
+    ) -> List[str]:
         if value:
             fields = [param.strip() for param in value.split(",")]
-            return self.remove_invalid_fields(queryset, fields)
+            return self.remove_invalid_fields(items, fields)
         return []
 
-    def remove_invalid_fields(self, queryset: QuerySet, fields: List[str]) -> List[str]:
-        valid_fields = [item for item in self.get_valid_fields(queryset)]
+    def remove_invalid_fields(
+        self, items: Union[QuerySet, List], fields: List[str]
+    ) -> List[str]:
+        valid_fields = [item for item in self.get_valid_fields(items)]
 
         def term_valid(term: str) -> bool:
             if term.startswith("-"):
@@ -92,15 +119,34 @@ def term_valid(term: str) -> bool:
 
         return [term for term in fields if term_valid(term)]
 
-    def get_valid_fields(self, queryset: QuerySet) -> List[str]:
+    def get_valid_fields(self, items: Union[QuerySet, List]) -> List[str]:
         valid_fields: List[str] = []
         if self.ordering_fields == "__all__":
-            valid_fields = [str(field.name) for field in queryset.model._meta.fields]
-            valid_fields += [str(key) for key in queryset.query.annotations]
+            if isinstance(items, QuerySet):  # type:ignore
+                valid_fields = self.get_all_valid_fields_from_queryset(items)
+            elif isinstance(items, list):
+                valid_fields = self.get_all_valid_fields_from_list(items)
         else:
             valid_fields = [item for item in self.ordering_fields]
         return valid_fields
 
+    def get_all_valid_fields_from_queryset(self, items: QuerySet) -> List[str]:
+        return [str(field.name) for field in items.model._meta.fields] + [
+            str(key) for key in items.query.annotations
+        ]
+
+    def get_all_valid_fields_from_list(self, items: List) -> List[str]:
+        if not items:
+            return []
+        item = items[0]
+        if isinstance(item, BaseModel):
+            return list(item.__fields__.keys())
+        if isinstance(item, dict):
+            return list(item.keys())
+        if hasattr(item, "_meta") and hasattr(item._meta, "fields"):
+            return [str(field.name) for field in item._meta.fields]
+        return []
+
 
 @overload
 def ordering() -> Callable[..., Any]:  # pragma: no cover
@@ -114,7 +160,9 @@ def ordering(
     ...
 
 
-def ordering(func_or_ordering_class: Any = NOT_SET, **ordering_params: Any) -> Callable:
+def ordering(
+    func_or_ordering_class: Any = NOT_SET, **ordering_params: Any
+) -> Callable[..., Any]:
     isfunction = inspect.isfunction(func_or_ordering_class)
     isnotset = func_or_ordering_class == NOT_SET
 
diff --git a/ninja_extra/searching.py b/ninja_extra/searching.py
@@ -1,15 +1,18 @@
 import inspect
 import operator
+import re
 from abc import ABC, abstractmethod
 from functools import reduce, wraps
 from typing import (
     TYPE_CHECKING,
     Any,
     Callable,
+    Dict,
     List,
     Optional,
     Tuple,
     Type,
+    Union,
     cast,
     overload,
 )
@@ -36,33 +39,58 @@
 ]
 
 
+def _istartswith(a: str, b: str) -> bool:
+    return a.startswith(b)
+
+
+def _isiexact(a: str, b: str) -> bool:
+    return a.lower() == b.lower()
+
+
+def _isiregex(a: str, b: str) -> bool:
+    return bool(re.search(b, a, re.IGNORECASE))
+
+
+def _isicontains(a: str, b: str) -> bool:
+    return b.lower() in a.lower()
+
+
 class SearchingBase(ABC):
     class Input(Schema):
         ...
 
-    lookup_prefixes = {
-        "^": "istartswith",
-        "=": "iexact",
-        "@": "search",
-        "$": "iregex",
-    }
     InputSource = Query(...)
 
     def __init__(self, *, pass_parameter: Optional[str] = None, **kwargs: Any) -> None:
         self.pass_parameter = pass_parameter
 
     @abstractmethod
-    def searching_queryset(self, queryset: QuerySet, searching_input: Any) -> QuerySet:
+    def searching_queryset(
+        self, items: Union[QuerySet, List], searching_input: Any
+    ) -> Union[QuerySet, List]:
         ...
 
 
 class Searching(SearchingBase):
     class Input(Schema):
         search: Optional[str] = Field(None)
 
+    lookup_prefixes = {
+        "^": "istartswith",
+        "=": "iexact",
+        "@": "search",
+        "$": "iregex",
+    }
+
+    lookup_prefixes_list = {
+        "^": _istartswith,
+        "=": _isiexact,
+        "$": _isiregex,
+    }
+
     def __init__(
         self,
-        search_fields: Optional[List[str]] = None,
+        search_fields: List[str] = [],
         pass_parameter: Optional[str] = None,
     ) -> None:
         self.search_fields = search_fields
@@ -76,23 +104,23 @@ class DynamicInput(Searching.Input):
         return DynamicInput
 
     def searching_queryset(
-        self, queryset: QuerySet, searching_input: Input
-    ) -> QuerySet:
+        self, items: Union[QuerySet, List], searching_input: Input
+    ) -> Union[QuerySet, List]:
 
         search_terms = self.get_search_terms(searching_input.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))
-        return queryset.filter(reduce(operator.and_, conditions))
+        if self.search_fields and search_terms:
+            if isinstance(items, QuerySet):  # type:ignore
+                conditions_queryset = self.construct_conditions_for_queryset(
+                    search_terms
+                )
+                return items.filter(reduce(operator.and_, conditions_queryset))
+            elif isinstance(items, list):
+                conditions_list = self.construct_conditions_for_list(search_terms)
+                return [
+                    item for item in items if self.filter_spec(item, conditions_list)
+                ]
+        return items
 
     def get_search_terms(self, value: Optional[str]) -> List[str]:
         if value:
@@ -109,6 +137,56 @@ def construct_search(self, field_name: str) -> str:
             lookup = "icontains"
         return LOOKUP_SEP.join([field_name, lookup])
 
+    def construct_conditions_for_queryset(self, search_terms: List[str]) -> List[Q]:
+        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))
+        return conditions
+
+    def construct_conditions_for_list(
+        self, search_terms: List[str]
+    ) -> Dict[str, List[Tuple[Callable, str]]]:
+        lookups = self.construct_search_for_list()
+        conditions: Dict[str, List[Tuple[Callable, str]]] = {
+            field_name: [] for field_name in lookups.keys()
+        }
+        for search_term in search_terms:
+            for field_name, lookup in lookups.items():
+                conditions[field_name].append((lookup, search_term))
+        return conditions
+
+    def construct_search_for_list(self) -> Dict[str, Callable]:
+        def get_lookup(prefix: str) -> Callable:
+            return self.lookup_prefixes_list.get(prefix, _isicontains)
+
+        return {
+            field_name[1:]
+            if (self.lookup_prefixes_list.get(field_name[0]))
+            else field_name: get_lookup(field_name[0])
+            for field_name in self.search_fields
+        }
+
+    def filter_spec(
+        self, item: Any, conditions: Dict[str, List[Tuple[Callable, str]]]
+    ) -> bool:
+        item_getter = (
+            operator.itemgetter if isinstance(item, dict) else operator.attrgetter
+        )
+        for field, lookup in conditions.items():
+            if not any(
+                [
+                    lookup_func(item_getter(field)(item), lookup_value)
+                    for lookup_func, lookup_value in lookup
+                ]
+            ):
+                return False
+        return True
+
 
 @overload
 def searching() -> Callable[..., Any]:  # pragma: no cover
@@ -124,7 +202,7 @@ def searching(
 
 def searching(
     func_or_searching_class: Any = NOT_SET, **searching_params: Any
-) -> Callable:
+) -> Callable[..., Any]:
     isfunction = inspect.isfunction(func_or_searching_class)
     isnotset = func_or_searching_class == NOT_SET
 
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -7,7 +7,7 @@ def pytest_configure(config):
     from django.conf import settings
 
     os.environ.setdefault("NINJA_SKIP_REGISTRY", "True")
-    os.environ.setdefault("DJANGO_ALLOW_ASYNC_UNSAFE", "True")
+
     settings.configure(
         ALLOWED_HOSTS=["*"],
         DEBUG_PROPAGATE_EXCEPTIONS=True,
diff --git a/tests/test_ordering.py b/tests/test_ordering.py
diff --git a/tests/test_searching.py b/tests/test_searching.py
