Merge pull request #455 from turicas/feature/validate-filter-data

Cria model forms dinâmicos para poder validar inputs de querystring

api/views.py

@@ -6,6 +6,8 @@
 from rest_framework.response import Response
 
 from api.serializers import DatasetDetailSerializer, DatasetSerializer, GenericSerializer
+from core.filters import parse_querystring
+from core.forms import get_table_dynamic_form
 from core.models import Dataset, Table
 from core.templatetags.utils import obfuscate
 
@@ -26,6 +28,11 @@ def retrieve(self, request, slug):
         return Response(serializer.data)
 
 
+class InvalidFiltersException(Exception):
+    def __init__(self, errors_list):
+        self.errors_list = errors_list
+
+
 class DatasetDataListView(ListAPIView):
 
     pagination_class = paginators.LargeTablePageNumberPagination
@@ -44,9 +51,16 @@ def get_queryset(self):
                 del querystring[pagination_key]
 
         Model = self.get_model_class()
-        queryset = Model.objects.filter_by_querystring(querystring)
+        query, search_query, order_by = parse_querystring(querystring)
 
-        return queryset
+        DynamicForm = get_table_dynamic_form(self.get_table())
+        filter_form = DynamicForm(data=query)
+        if filter_form.is_valid():
+            query = {k: v for k, v in filter_form.cleaned_data.items() if v != ""}
+        else:
+            raise InvalidFiltersException(filter_form.errors)
+
+        return Model.objects.composed_query(query, search_query, order_by)
 
     def get_serializer_class(self):
         table = self.get_table()
@@ -72,6 +86,12 @@ def get_serializer(self, *args, **kwargs):
 
         return super().get_serializer(*args, **kwargs)
 
+    def handle_exception(self, exc):
+        if isinstance(exc, InvalidFiltersException):
+            return Response(exc.errors_list, status=400)
+        else:
+            return super().handle_exception(exc)
+
 
 dataset_list = DatasetViewSet.as_view({"get": "list"})
 dataset_detail = DatasetViewSet.as_view({"get": "retrieve"}, lookup_field="slug")

core/filters.py

@@ -8,6 +8,15 @@ def clean_value(key, value):
     return key, value
 
 
+def parse_querystring(querystring):
+    query = querystring.copy()
+    order_by = query.pop("order-by", [""])
+    order_by = [field.strip().lower() for field in order_by[0].split(",") if field.strip()]
+    search_query = query.pop("search", [""])[0]
+    query = {key: value for key, value in query.items() if value}
+    return query, search_query, order_by
+
+
 class DynamicModelFilterProcessor:
     def __init__(self, filtering: dict, allowed_filters: list):
         self.filtering = filtering

core/forms.py

@@ -104,3 +104,20 @@ class ContactForm(forms.Form):
 
 class DatasetSearchForm(forms.Form):
     search = forms.CharField(label="Titulo ou Descrição")
+
+
+def get_table_dynamic_form(table, cache=True):
+    def config_dynamic_filter(model_field):
+        dynamic_field = table.get_field(model_field.name)
+        kwargs = {"required": False, "label": dynamic_field.title}
+        field_factory = model_field.formfield
+
+        if dynamic_field.has_choices:
+            kwargs["choices"] = [("", "Todos")] + [(c, c) for c in dynamic_field.choices["data"]]
+            field_factory = forms.ChoiceField
+
+        return field_factory(**kwargs)
+
+    model = table.get_model(cache=cache)
+    fields = model.extra["filtering"]
+    return forms.modelform_factory(model, fields=fields, formfield_callback=config_dynamic_filter)

core/models.py

@@ -113,18 +113,6 @@ def apply_ordering(self, query):
 
         return qs
 
-    def filter_by_querystring(self, querystring):
-        query, search_query, order_by = self.parse_querystring(querystring)
-        return self.composed_query(query, search_query, order_by)
-
-    def parse_querystring(self, querystring):
-        query = querystring.copy()
-        order_by = query.pop("order-by", [""])
-        order_by = [field.strip().lower() for field in order_by[0].split(",") if field.strip()]
-        search_query = query.pop("search", [""])[0]
-        query = {key: value for key, value in query.items() if value}
-        return query, search_query, order_by
-
     def composed_query(self, filter_query=None, search_query=None, order_by=None):
         qs = self
         if search_query:
@@ -377,6 +365,9 @@ def get_dynamic_model_mixins(self):
         custom_mixins = [] if not self.dynamic_table_config else self.dynamic_table_config.get_model_mixins()
         return custom_mixins + mixins
 
+    def get_field(self, name):
+        return self.fields.get(name=name)
+
     def get_model(self, cache=True, data_table=None):
         # TODO: the current dynamic model registry is handled by Brasil.IO's
         # code but it needs to be delegated to dynamic_models.
@@ -423,6 +414,7 @@ def get_model(self, cache=True, data_table=None):
             "filtering": filtering,
             "ordering": ordering,
             "search": search,
+            "table": self,
         }
         DYNAMIC_MODEL_REGISTRY[cache_key] = Model
         return Model

core/templates/core/dataset-detail.html

@@ -82,23 +82,15 @@ <h5>Filtros</h5>
                 <input type="text" id="search" name="search" value="{{ query_dict.search|default:'' }}">
               </div>
 
-              {% for field in fields %}
-              {% if field.frontend_filter %}{% with value=query_dict|getplainattribute:field|default:'' %}
+              {% for field in filter_form %}
               <div class="input-field col s6">
-                <label class="active" for="{{ field.name }}">{{ field.title }}</label>
-              {% if field.has_choices %}
-                <select name="{{ field.name }}">
-                  <option value="" {% if value == "" %} selected{% endif %}>Todos</option>
-                  {% for choice in field.choices.data %}
-                  <option value="{{ choice }}"{% if value == choice %} selected{% endif %}>{% if choice == 'None' %}(vazio){% else %}{{ choice }}{% endif %}</option>
-                  {% endfor %}
-                </select>
-                {% else %}
-                <input type="text" name="{{ field.name }}" value="{{ value }}">
-              {% endif %}
+                <label class="active" for="{{ field.name }}">{{ field.label }}</label>
+                {{ field }}
+                {% if field.errors %}
+                  <span class="filter-error">{{ field.errors.as_text }}</span>
+                {% endif %}
               </div>
-            {% endwith %}{% endif %}
-            {% endfor %}
+              {% endfor %}
 
             </div>
             <div class="row">

core/templatetags/utils.py

@@ -25,11 +25,6 @@ def getattribute(obj, field):
     return _getattr(obj, field, should_obfuscate=True)
 
 
-@register.filter(name="getplainattribute")
-def getplainattribute(obj, field):
-    return _getattr(obj, field, should_obfuscate=False)
-
-
 @register.filter(name="render")
 def render(template_text, obj):
     template_text = "{% load utils %}" + template_text  # inception

core/tests/test_forms.py

@@ -0,0 +1,76 @@
+from django.forms import ChoiceField
+
+from core.forms import get_table_dynamic_form
+from core.tests.utils import BaseTestCaseWithSampleDataset
+
+
+class DynamicModelFormTests(BaseTestCaseWithSampleDataset):
+    DATASET_SLUG = "sample"
+    TABLE_NAME = "sample_table"
+    FIELDS_KWARGS = [
+        {"name": "name", "options": {"max_length": 50}, "type": "text", "null": False},
+        {"name": "uf", "options": {"max_length": 2}, "type": "text", "null": False},
+        {"name": "city", "options": {"max_length": 50}, "type": "text", "null": False},
+    ]
+
+    def setUp(self):
+        self.table.filtering = []
+        self.table.save()
+
+    def get_model_field(self, name):
+        return [f for f in self.TableModel._meta.fields if f.name == name][0]
+
+    def test_table_without_filters_gets_empty_form(self):
+        DynamicFormClasss = get_table_dynamic_form(self.table, cache=False)
+        form = DynamicFormClasss()
+        assert 0 == len(form.fields)
+
+    def test_generate_form_based_in_table_filtering(self):
+        self.table.filtering = ["uf", "city"]
+        self.table.save()
+
+        DynamicFormClasss = get_table_dynamic_form(self.table, cache=False)
+        form = DynamicFormClasss()
+
+        assert "uf" in form.fields
+        assert "city" in form.fields
+        assert isinstance(form.fields["uf"], type(self.get_model_field("uf").formfield()))
+        assert isinstance(form.fields["city"], type(self.get_model_field("city").formfield()))
+
+    def test_filter_form_does_not_invalidate_if_no_data(self):
+        self.table.filtering = ["uf", "city"]
+        self.table.save()
+
+        DynamicFormClasss = get_table_dynamic_form(self.table, cache=False)
+        form = DynamicFormClasss(data={})
+
+        assert form.is_valid()
+        assert {"uf": "", "city": ""} == form.cleaned_data
+
+    def test_validate_form_against_field_choices(self):
+        self.table.filtering = ["uf", "city"]
+        self.table.save()
+        uf_field = self.table.get_field("uf")
+        uf_field.has_choices = True
+        uf_field.choices = {"data": ["RJ", "SP", "MG"]}
+        uf_field.save()
+        city_field = self.table.get_field("city")
+        city_field.has_choices = True
+        city_field.choices = {"data": ["Rio de Janeiro", "São Pauyo", "Belo Horizonte"]}
+        city_field.save()
+
+        # valid form
+        DynamicFormClasss = get_table_dynamic_form(self.table, cache=False)
+        form = DynamicFormClasss(data={"uf": "RJ", "city": "Rio de Janeiro"})
+        assert isinstance(form.fields["uf"], ChoiceField)
+        assert isinstance(form.fields["city"], ChoiceField)
+        assert form.is_valid()
+        assert {"uf": "RJ", "city": "Rio de Janeiro"} == form.cleaned_data
+
+        # invalid form
+        DynamicFormClasss = get_table_dynamic_form(self.table, cache=False)
+        form = DynamicFormClasss(data={"uf": "XXX", "city": "Rio de Janeiro"})
+        assert not form.is_valid()
+        assert "uf" in form.errors
+        assert "city" not in form.errors
+        assert {"city": "Rio de Janeiro"} == form.cleaned_data

core/views.py

@@ -10,7 +10,8 @@
 from django.shortcuts import get_object_or_404, redirect, render
 from django.urls import reverse
 
-from core.forms import ContactForm, DatasetSearchForm
+from core.filters import parse_querystring
+from core.forms import ContactForm, DatasetSearchForm, get_table_dynamic_form
 from core.middlewares import disable_non_logged_user_cache
 from core.models import Dataset, Table
 from core.templatetags.utils import obfuscate
@@ -146,10 +147,17 @@ def dataset_detail(request, slug, tablename=""):
     items_per_page = min(items_per_page, 1000)
 
     version = dataset.version_set.order_by("-order").first()
-    fields = table.fields
 
     TableModel = table.get_model()
-    query, search_query, order_by = TableModel.objects.parse_querystring(querystring)
+    query, search_query, order_by = parse_querystring(querystring)
+
+    DynamicForm = get_table_dynamic_form(table)
+    filter_form = DynamicForm(data=query)
+    if filter_form.is_valid():
+        query = {k: v for k, v in filter_form.cleaned_data.items() if v != ""}
+    else:
+        query = {}
+
     all_data = TableModel.objects.composed_query(query, search_query, order_by)
 
     if download_csv:
@@ -172,7 +180,7 @@ def dataset_detail(request, slug, tablename=""):
         filename = "{}-{}.csv".format(slug, uuid.uuid4().hex)
         pseudo_buffer = Echo()
         writer = csv.writer(pseudo_buffer, dialect=csv.excel)
-        csv_rows = queryset_to_csv(all_data, fields)
+        csv_rows = queryset_to_csv(all_data, table.fields)
         response = StreamingHttpResponse(
             (writer.writerow(row) for row in csv_rows), content_type="text/csv;charset=UTF-8",
         )
@@ -190,7 +198,7 @@ def dataset_detail(request, slug, tablename=""):
     context = {
         "data": data,
         "dataset": dataset,
-        "fields": fields,
+        "filter_form": filter_form,
         "max_export_rows": settings.CSV_EXPORT_MAX_ROWS,
         "query_dict": querystring,
         "querystring": querystring.urlencode(),
@@ -200,7 +208,11 @@ def dataset_detail(request, slug, tablename=""):
         "total_count": all_data.count(),
         "version": version,
     }
-    return render(request, "core/dataset-detail.html", context)
+
+    status = 200
+    if filter_form.errors:
+        status = 400
+    return render(request, "core/dataset-detail.html", context, status=status)
 
 
 def dataset_suggestion(request):

static/css/base.css

@@ -191,4 +191,9 @@ a {
 
 .error-container p{
   font-size: 1.4rem;
-}
+}
+
+.filter-error {
+  color: #ee6e73;
+  font-size: 1rem;
+}