[Feature] Basic framework for datatables server rendering

allianceauth/framework/datatables.py

@@ -0,0 +1,216 @@
+from collections import defaultdict
+import re
+from typing import List
+
+from django.db.models import Model, Q
+from django.http import HttpRequest, JsonResponse
+from django.template import Context, Template
+from django.template.loader import render_to_string
+from django.views import View
+
+from allianceauth.services.hooks import get_extension_logger
+
+
+logger = get_extension_logger(__name__)
+
+
+class nested_param_dict(dict):
+    """
+    Helper to create infinite depth default dicts for setting from params
+    """
+    def __setitem__(self, item, value):
+        if "." in item:
+            head, path = item.split(".", 1)
+            try:
+                head = int(head)
+            except ValueError:
+                pass
+            obj = self.setdefault(head, nested_param_dict())
+            obj[path] = value
+        else:
+            super().__setitem__(item, value)
+
+
+def defaultdict_to_dict(d):
+    """
+    Helper to convert default dict back to dict
+    """
+    if isinstance(d, defaultdict):
+        d = {k: defaultdict_to_dict(v) for k, v in d.items()}
+    return d
+
+
+class DataTablesView(View):
+
+    model: Model = None
+    columns: List[tuple] = []
+
+    def get_model_qs(self, request: HttpRequest, *args, **kwargs):
+        return self.model.objects
+
+    def filter_qs(self, table_conf: dict):
+        # Search
+        filter_qs = Q()
+        for id, c in table_conf["columns"].items():
+            _c = self.columns[int(id)][0]
+            if c.get("searchable", False) and len(_c) > 0:
+                if c.get("columnControl", False):
+                    _sv = str(c["columnControl"]["search"]["value"])
+                    """contains, equal, ends, starts, empty"""
+                    _logic = str(c["columnControl"]["search"]["logic"])
+                    """text, date, num"""
+                    _type = str(c["columnControl"]["search"]["type"])
+                    if _type == "text":
+                        if _logic == "empty":
+                            filter_qs &= Q(**{f'{_c}': ""})
+                        elif len(_sv) > 0:
+                            if _logic == "contains":
+                                filter_qs &= Q(**{f'{_c}__icontains': _sv})
+                            elif _logic == "starts":
+                                filter_qs &= Q(**{f'{_c}__istartswith': _sv})
+                            elif _logic == "ends":
+                                filter_qs &= Q(**{f'{_c}__iendswith': _sv})
+                            elif _logic == "equal":
+                                filter_qs &= Q(**{f'{_c}': _sv})
+                    elif _type == "num":
+                        if _logic == "empty":
+                            filter_qs &= Q(**{f'{_c}__isnull': True})
+                        elif len(_sv) > 0:
+                            try:
+                                if _logic == "greater":
+                                    filter_qs &= Q(**{f'{_c}__gt': float(_sv)})
+                                elif _logic == "less":
+                                    filter_qs &= Q(**{f'{_c}__lt': float(_sv)})
+                                elif _logic == "greaterOrEqual":
+                                    filter_qs &= Q(**{f'{_c}__gte': float(_sv)})
+                                elif _logic == "lessOrEqual":
+                                    filter_qs &= Q(**{f'{_c}__lte': float(_sv)})
+                                elif _logic == "equal":
+                                    filter_qs &= Q(**{f'{_c}': float(_sv)})
+                            except ValueError:
+                                pass
+                else:
+                    _sv = str(c["search"]["value"])
+                    if len(_sv) > 0:
+                        if c["search"]["regex"]:
+                            filter_qs |= Q(**{f'{_c}__iregex': _sv})
+                        else:
+                            filter_qs |= Q(**{f'{_c}__icontains': _sv})
+                _gsv = str(table_conf["search"]["value"])
+                if len(_gsv) > 0:
+                    filter_qs |= Q(**{f'{_c}__icontains': _gsv})
+        return filter_qs
+
+    def except_qs(self, table_conf: dict):
+        # Search
+        except_qs = Q()
+        for id, c in table_conf["columns"].items():
+            _c = self.columns[int(id)][0]
+            if c.get("searchable", False) and len(_c) > 0:
+                if c.get("columnControl", False):
+                    _sv = str(c["columnControl"]["search"]["value"])
+                    """notContains, notEqual, notEmpty"""
+                    _logic = str(c["columnControl"]["search"]["logic"])
+                    """text, date, num"""
+                    _type = str(c["columnControl"]["search"]["type"])
+                    if _type == "text":
+                        if _logic == "notEmpty":
+                                except_qs |= Q(**{f'{_c}': ""})
+                        elif len(_sv) > 0:
+                            if _logic == "notContains":
+                                except_qs |= Q(**{f'{_c}__icontains': _sv})
+                            elif _logic == "notEqual":
+                                except_qs |= Q(**{f'{_c}': _sv})
+                    elif _type == "num":
+                        if _logic == "notEmpty":
+                                except_qs |= Q(**{f'{_c}__isnull': False})
+                        elif len(_sv) > 0:
+                            if _logic == "notEqual":
+                                try:
+                                    except_qs |= Q(**{f'{_c}': float(_sv)})
+                                except ValueError:
+                                    pass
+        return except_qs
+
+    def get_table_config(self, get: dict):
+        _cols = nested_param_dict()
+        for c, v in get.items():
+            _keys = [_k for _k in c.replace("]", "").split("[")]
+            _v = v
+            if v in ["true", "false"]:
+                _v = _v == "true"
+            else:
+                try:
+                    _v = int(_v)
+                except ValueError:
+                    pass # not an integer
+            _cols[".".join(_keys)] = _v
+        return defaultdict_to_dict(_cols)
+
+    def get_order(self, table_conf: dict):
+        order = []
+        for oc, od in table_conf.get("order", {}).items():
+            _c = table_conf["columns"][od["column"]]
+            if _c["orderable"]:
+                if od["dir"] == "desc":
+                    order.append("-" + self.columns[int(od["column"])][0])
+                else:
+                    order.append(self.columns[int(od["column"])][0])
+        return order
+
+    def render_template(self, request: HttpRequest, template: str, ctx: dict):
+        if "{{" in template:
+            _template = Template(template)
+            return _template.render(Context(ctx))
+        else:
+            return render_to_string(
+                template,
+                ctx,
+                request
+            )
+
+    def get(self, request: HttpRequest, *args, **kwargs):
+        table_conf = self.get_table_config(request.GET)
+        draw = int(table_conf["draw"])
+        start = int(table_conf["start"])
+        length = int(table_conf["length"])
+        if length <= 0:
+            logger.warning(
+                "Using no pagination is not recommended for server side rendered datatables"
+            )
+        limit = start + length
+
+
+        # Build response rows
+        items = []
+        qs = self.get_model_qs(
+            request,
+            *args,
+            **kwargs
+        ).filter(
+            self.filter_qs(table_conf)
+        ).exclude(
+            self.except_qs(table_conf)
+        ).order_by(
+            *self.get_order(table_conf)
+        )
+
+        # build output
+        if length > 0:
+            qs = qs[start:limit]
+
+        for row in qs:
+            ctx = {"row": row}
+            row = []
+            for t in self.columns:
+                row.append(self.render_template(request, t[1], ctx))
+            items.append(row)
+
+        # Build our output dict
+        datatables_data = {}
+        datatables_data['draw'] = draw
+        datatables_data['recordsTotal'] = self.get_model_qs(request, *args, **kwargs).all().count()
+        datatables_data['recordsFiltered'] = qs.count()
+        datatables_data['data'] = items
+
+        return JsonResponse(datatables_data)

allianceauth/framework/tests/test_datatables.py

@@ -0,0 +1,278 @@
+"""
+Test sentinel user
+"""
+
+import json
+import re
+
+# Django
+from allianceauth.tests.auth_utils import AuthUtils
+from django.test import RequestFactory, TestCase
+from django.http import HttpRequest
+# Alliance Auth
+from allianceauth.framework.datatables import DataTablesView
+from allianceauth.eveonline.models import EveCharacter
+
+class TestView(DataTablesView):
+    model=EveCharacter
+    columns = [
+        ("", "{{ row.character_id }}"),
+        ("character_name", "{{ row.character_name }}"),
+        ("corporation_name", "{{ row.corporation_name }}"),
+        ("alliance_name", "{{ row.alliance_name }}"),
+    ]
+
+class TestDataTables(TestCase):
+    def setUp(self):
+        self.get_params = {
+            'draw': '1',
+            'columns[0][data]': '0',
+            'columns[0][name]': '',
+            'columns[0][searchable]': 'false',
+            'columns[0][orderable]': 'false',
+            'columns[0][search][value]': '',
+            'columns[0][search][regex]': 'false',
+            'columns[1][data]': '1',
+            'columns[1][name]': '',
+            'columns[1][searchable]': 'true',
+            'columns[1][orderable]': 'true',
+            'columns[1][search][value]': '',
+            'columns[1][search][regex]': 'false',
+            'columns[2][data]': '2',
+            'columns[2][name]': '',
+            'columns[2][searchable]': 'true',
+            'columns[2][orderable]': 'false',
+            'columns[2][search][value]': '',
+            'columns[2][search][regex]': 'false',
+            'columns[3][data]': '3',
+            'columns[3][name]': '',
+            'columns[3][searchable]': 'true',
+            'columns[3][orderable]': 'true',
+            'columns[3][search][value]': '',
+            'columns[3][search][regex]': 'false',
+            'order[0][column]': '1',
+            'order[0][dir]': 'asc',
+            'start': '0',
+            'length': '10',
+            'search[value]': '',
+            'search[regex]': 'false',
+            '_': '123456789'
+        }
+
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        """
+        Set up eve models
+        """
+
+        super().setUpClass()
+        cls.factory = RequestFactory()
+
+        cls.user = AuthUtils.create_user("bruce_wayne")
+        cls.user.is_superuser = True
+        cls.user.save()
+
+        EveCharacter.objects.all().delete()
+        for i in range(1,16):
+            EveCharacter.objects.create(
+                character_id=1000+i,
+                character_name=f"{1000+i} - Test Character - {1000+i}",
+                corporation_id=2000+i,
+                corporation_name=f"{2000+i} - Test Corporation",
+            )
+
+        for i in range(16,21):
+            EveCharacter.objects.create(
+                character_id=1000+i,
+                character_name=f"{1000+i} - Test Character - {1000+i}",
+                corporation_id=2000+i,
+                corporation_name=f"{2000+i} - Test Corporation",
+                alliance_id=3000+i,
+                alliance_name=f"{3000+i} - Test Alliance",
+            )
+
+
+    def test_view_default(self):
+        self.client.force_login(self.user)
+        request = self.factory.get('/fake-url/', data=self.get_params)
+        response = TestView()
+        response.setup(request)
+        data = json.loads(response.get(request).content)["data"]
+        self.assertEqual(data[0][0], "1001")
+        self.assertEqual(data[9][0], "1010")
+
+    def test_view_reverse_sort(self):
+        self.get_params["order[0][dir]"] = "desc"
+        self.client.force_login(self.user)
+        request = self.factory.get('/fake-url/', data=self.get_params)
+        response = TestView()
+        response.setup(request)
+        data = json.loads(response.get(request).content)["data"]
+        self.assertEqual(data[0][0], "1020")
+        self.assertEqual(data[9][0], "1011")
+
+    def test_view_no_sort(self):
+        self.get_params.pop("order[0][column]")
+        self.get_params.pop("order[0][dir]")
+        self.client.force_login(self.user)
+        request = self.factory.get('/fake-url/', data=self.get_params)
+        response = TestView()
+        response.setup(request)
+        data = json.loads(response.get(request).content)["data"]
+        self.assertEqual(data[0][0], "1001")
+        self.assertEqual(data[9][0], "1010")
+
+    def test_view_non_sortable_sort(self):
+        self.get_params["order[0][dir]"] = "desc"
+        self.get_params["order[0][column]"] = "0"
+        self.client.force_login(self.user)
+        request = self.factory.get('/fake-url/', data=self.get_params)
+        response = TestView()
+        response.setup(request)
+        data = json.loads(response.get(request).content)["data"]
+        self.assertEqual(data[0][0], "1001")
+        self.assertEqual(data[9][0], "1010")
+
+    def test_view_20_rows(self):
+        self.get_params["length"] = "20"
+        self.client.force_login(self.user)
+        request = self.factory.get('/fake-url/', data=self.get_params)
+        response = TestView()
+        response.setup(request)
+        data = json.loads(response.get(request).content)["data"]
+        self.assertEqual(data[0][0], "1001")
+        self.assertEqual(data[19][0], "1020")
+
+    def test_view_global_search(self):
+        self.get_params["search[value]"] = "1020"
+        self.client.force_login(self.user)
+        request = self.factory.get('/fake-url/', data=self.get_params)
+        response = TestView()
+        response.setup(request)
+        data = json.loads(response.get(request).content)["data"]
+        self.assertEqual(len(data), 1)
+        self.assertEqual(data[0][0], "1020")
+
+    def test_view_col_1_search(self):
+        self.get_params["columns[1][search][value]"] = "1020"
+        self.client.force_login(self.user)
+        request = self.factory.get('/fake-url/', data=self.get_params)
+        response = TestView()
+        response.setup(request)
+        data = json.loads(response.get(request).content)["data"]
+        self.assertEqual(len(data), 1)
+        self.assertEqual(data[0][0], "1020")
+
+    def test_view_col_1_search_empty(self):
+        self.get_params["columns[1][search][value]"] = "zzz"
+
+        self.client.force_login(self.user)
+        request = self.factory.get('/fake-url/', data=self.get_params)
+        response = TestView()
+        response.setup(request)
+        data = json.loads(response.get(request).content)["data"]
+        self.assertEqual(len(data), 0)
+
+    def test_view_cc_3_search_empty(self):
+        self.get_params["columns[3][columnControl][search][value]"] = ""
+        self.get_params["columns[3][columnControl][search][logic]"] = "empty"
+        self.get_params["columns[3][columnControl][search][type]"] = "text"
+        self.get_params["length"] = "20"
+
+        self.client.force_login(self.user)
+        request = self.factory.get('/fake-url/', data=self.get_params)
+        response = TestView()
+        response.setup(request)
+        data = json.loads(response.get(request).content)["data"]
+        self.assertEqual(len(data), 15)
+
+    def test_view_cc_3_search_not_empty(self):
+        self.get_params["columns[3][columnControl][search][value]"] = ""
+        self.get_params["columns[3][columnControl][search][logic]"] = "notEmpty"
+        self.get_params["columns[3][columnControl][search][type]"] = "text"
+        self.client.force_login(self.user)
+        request = self.factory.get('/fake-url/', data=self.get_params)
+        response = TestView()
+        response.setup(request)
+        data = json.loads(response.get(request).content)["data"]
+        self.assertEqual(len(data), 5)
+
+    def test_view_cc_1_search_ends_with(self):
+        self.get_params["columns[1][columnControl][search][value]"] = "9"
+        self.get_params["columns[1][columnControl][search][logic]"] = "ends"
+        self.get_params["columns[1][columnControl][search][type]"] = "text"
+        self.client.force_login(self.user)
+        request = self.factory.get('/fake-url/', data=self.get_params)
+        response = TestView()
+        response.setup(request)
+        data = json.loads(response.get(request).content)["data"]
+        self.assertEqual(len(data), 2)
+
+    def test_view_cc_1_search_starts_with(self):
+        self.get_params["columns[1][columnControl][search][value]"] = "1009"
+        self.get_params["columns[1][columnControl][search][logic]"] = "starts"
+        self.get_params["columns[1][columnControl][search][type]"] = "text"
+        self.client.force_login(self.user)
+        request = self.factory.get('/fake-url/', data=self.get_params)
+        response = TestView()
+        response.setup(request)
+        data = json.loads(response.get(request).content)["data"]
+        self.assertEqual(len(data), 1)
+
+    def test_view_cc_1_search_not_contains(self):
+        self.get_params["columns[1][columnControl][search][value]"] = "100"
+        self.get_params["columns[1][columnControl][search][logic]"] = "notContains"
+        self.get_params["columns[1][columnControl][search][type]"] = "text"
+        self.get_params["length"] = "20"
+        self.client.force_login(self.user)
+        request = self.factory.get('/fake-url/', data=self.get_params)
+        response = TestView()
+        response.setup(request)
+        data = json.loads(response.get(request).content)["data"]
+        self.assertEqual(len(data), 11)
+
+    def test_view_cc_1_search_contains(self):
+        self.get_params["columns[1][columnControl][search][value]"] = "100"
+        self.get_params["columns[1][columnControl][search][logic]"] = "contains"
+        self.get_params["columns[1][columnControl][search][type]"] = "text"
+        self.get_params["length"] = "20"
+        self.client.force_login(self.user)
+        request = self.factory.get('/fake-url/', data=self.get_params)
+        response = TestView()
+        response.setup(request)
+        data = json.loads(response.get(request).content)["data"]
+        self.assertEqual(len(data), 9)
+
+    def test_view_cc_1_search_equal(self):
+        self.get_params["columns[1][columnControl][search][value]"] = "1001 - Test Character - 1001"
+        self.get_params["columns[1][columnControl][search][logic]"] = "equal"
+        self.get_params["columns[1][columnControl][search][type]"] = "text"
+        self.get_params["length"] = "20"
+        self.client.force_login(self.user)
+        request = self.factory.get('/fake-url/', data=self.get_params)
+        response = TestView()
+        response.setup(request)
+        data = json.loads(response.get(request).content)["data"]
+        self.assertEqual(len(data), 1)
+
+    def test_view_cc_1_search_not_equal(self):
+        self.get_params["columns[1][columnControl][search][value]"] = "1001 - Test Character - 1001"
+        self.get_params["columns[1][columnControl][search][logic]"] = "notEqual"
+        self.get_params["columns[1][columnControl][search][type]"] = "text"
+        self.get_params["length"] = "20"
+        self.client.force_login(self.user)
+        request = self.factory.get('/fake-url/', data=self.get_params)
+        response = TestView()
+        response.setup(request)
+        data = json.loads(response.get(request).content)["data"]
+        self.assertEqual(len(data), 19)
+
+    def test_view_cc_no_pagination(self):
+        self.get_params["length"] = "-1"
+        self.client.force_login(self.user)
+        request = self.factory.get('/fake-url/', data=self.get_params)
+        response = TestView()
+        response.setup(request)
+        data = json.loads(response.get(request).content)["data"]
+        self.assertEqual(len(data), 20)