Coverage for apis_core / generic / views.py: 81%
356 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-27 05:15 +0000
1import logging
2import traceback
3from collections import namedtuple
4from copy import copy
6from crispy_forms.layout import Field
7from dal import autocomplete
8from django import http
9from django.conf import settings
10from django.contrib import messages
11from django.contrib.auth.mixins import PermissionRequiredMixin
12from django.contrib.contenttypes.models import ContentType
13from django.contrib.messages.views import SuccessMessageMixin
14from django.core.exceptions import ImproperlyConfigured, ValidationError
15from django.core.validators import URLValidator
16from django.db import transaction
17from django.db.models.fields.related import ManyToManyRel
18from django.forms import modelform_factory
19from django.forms.utils import pretty_name
20from django.http import QueryDict
21from django.shortcuts import get_object_or_404, redirect
22from django.template.exceptions import TemplateDoesNotExist
23from django.template.loader import select_template
24from django.urls import reverse
25from django.utils.text import capfirst
26from django.views import View
27from django.views.generic import DetailView
28from django.views.generic.base import TemplateView
29from django.views.generic.edit import CreateView, DeleteView, FormView, UpdateView
30from django_filters.filterset import filterset_factory
31from django_filters.views import FilterView
32from django_tables2 import SingleTableMixin
33from django_tables2.columns import library
34from django_tables2.export.views import ExportMixin
35from django_tables2.tables import table_factory
37from apis_core.generic.utils import get_autocomplete_data_and_normalized_uri
38from apis_core.uris.models import Uri
40from .filtersets import GenericFilterSet
41from .forms import (
42 ColumnsSelectorForm,
43 GenericEnrichForm,
44 GenericImportForm,
45 GenericMergeWithForm,
46 GenericModelForm,
47 GenericSelectMergeOrEnrichForm,
48)
49from .helpers import (
50 first_member_match,
51 generate_search_filter,
52 module_paths,
53 permission_fullname,
54 template_names_via_mro,
55)
56from .tables import GenericTable
58logger = logging.getLogger(__name__)
61class Overview(TemplateView):
62 template_name = "generic/overview.html"
65class GenericModelPermissionRequiredMixin(PermissionRequiredMixin):
66 """
67 Verify that the current user has the required permission for this model.
68 The model overrides the `PermissionRequiredMixin.get_permission_required`
69 method to generate the required permission name on the fly, based on a
70 verb (`permission_action_required`) and the model this view act upon.
71 This allows us to set `permission_action_required` simply to `add`, or
72 `view` and reuse the mixin for views that work with different models.
73 In addition, for the views that have `permission_action_required` set to
74 `view`, it check if there is the global setting `APIS_ANON_VIEWS_ALLOWED`
75 set to `True`, which permits anonymouse users access to the view.
76 """
78 def get_permission_required(self):
79 if not hasattr(self, "model"):
80 raise ImproperlyConfigured(
81 f"{self.__class__.__name__} is missing the model attribute"
82 )
83 if getattr(self, "permission_action_required", None) == "view" and getattr(
84 settings, "APIS_ANON_VIEWS_ALLOWED", False
85 ):
86 return []
87 if hasattr(self, "permission_action_required"):
88 return [permission_fullname(self.permission_action_required, self.model)]
89 return []
92class GenericModelMixin:
93 """
94 A mixin providing the common functionality for all the views working
95 with `generic` models - that is models that are accessed via the
96 contenttype framework (using `app_label.model`).
97 It sets the `.model` of the view and generates a list of possible template
98 names (based on the MRO of the model).
99 If the view has a `permission_action_required` attribute, this is used
100 to set the permission required to access the view for this specific model.
101 """
103 def setup(self, *args, **kwargs):
104 super().setup(*args, **kwargs)
105 if contenttype := kwargs.get("contenttype"):
106 self.model = contenttype.model_class()
107 self.queryset = self.model.objects.all()
109 def get_template_names(self):
110 template_names = []
111 if hasattr(super(), "get_template_names"):
112 # Some parent classes come with custom template_names,
113 # some need a `.template_name` attribute set. For the
114 # latter ones we handle the missing `.template_name`
115 # gracefully
116 try:
117 template_names = super().get_template_names()
118 except ImproperlyConfigured:
119 pass
120 suffix = ".html"
121 if hasattr(self, "template_name_suffix"):
122 suffix = self.template_name_suffix + ".html"
123 additional_templates = template_names_via_mro(self.model, suffix=suffix)
124 template_names += filter(
125 lambda template: template not in template_names, additional_templates
126 )
127 return template_names
130class List(
131 GenericModelMixin,
132 GenericModelPermissionRequiredMixin,
133 ExportMixin,
134 SingleTableMixin,
135 FilterView,
136):
137 """
138 List view for a generic model.
139 Access requires the `<model>_view` permission.
140 It is based on django-filters FilterView and django-tables SingleTableMixin.
141 The table class is overridden by the first match from
142 the `first_member_match` helper.
143 The filterset class is overridden by the first match from
144 the `first_member_match` helper.
145 The queryset is overridden by the first match from
146 the `first_member_match` helper.
147 """
149 template_name_suffix = "_list"
150 permission_action_required = "view"
152 def setup(self, *args, **kwargs):
153 super().setup(*args, **kwargs)
154 content_type = ContentType.objects.get_for_model(self.model)
155 self.cookie_name = f"{content_type.app_label}.{content_type.model}-list"
156 cookie = QueryDict(self.request.COOKIES.get(self.cookie_name, ""))
157 get = self.request.GET.copy()
158 for prefix in ["filterset", "choices"]:
159 id_ = f"{prefix}-remember"
160 passed_prefix = any([prefix in key for key in self.request.GET.keys()])
161 use_cookie = cookie.get(id_, False) and not passed_prefix
162 if use_cookie:
163 for key in [key for key in cookie if key.startswith(prefix)]:
164 get.setlist(key, cookie.getlist(key))
165 if "sort" not in get.keys() and "sort" in cookie.keys():
166 get["sort"] = cookie.get("sort")
167 self.request.GET = get
169 def get(self, request, *args, **kwargs):
170 response = super().get(request, *args, **kwargs)
171 response.set_cookie(self.cookie_name, self.request.GET.urlencode())
172 return response
174 def get_table_class(self):
175 table_modules = module_paths(self.model, path="tables", suffix="Table")
176 table_class = first_member_match(table_modules, GenericTable)
177 return table_factory(self.model, table_class)
179 export_formats = getattr(settings, "EXPORT_FORMATS", ["csv", "json"])
181 def get_export_filename(self, extension):
182 table_class = self.get_table_class()
183 if hasattr(table_class, "export_filename"):
184 return f"{table_class.export_filename}.{extension}"
186 return super().get_export_filename(extension)
188 def get_table_kwargs(self):
189 kwargs = super().get_table_kwargs()
191 selected_columns = self.request.GET.getlist("choices-columns", [])
192 modelfields = self.model._meta.get_fields()
193 # if the form was submitted, we look at the selected
194 # columns and exclude all columns that are not part of that list
195 if self.request.GET and "choices-columns" in self.request.GET:
196 columns_exclude = self.get_filterset_class().Meta.form.columns_exclude
197 other_columns = [
198 name for (name, field) in self._get_columns_choices(columns_exclude)
199 ]
200 kwargs["exclude"] = [
201 field for field in other_columns if field not in selected_columns
202 ]
204 # now we look at the selected columns and
205 # add all modelfields and annotated fields that
206 # are part of the selected columns to the extra_columns
207 annotationfields = list()
208 for key, value in self.object_list.query.annotations.items():
209 # we have to use copy, so we don't edit the original field
210 fake_field = copy(getattr(value, "field", value.output_field))
211 setattr(fake_field, "name", key)
212 annotationfields.append(fake_field)
213 extra_fields = list(
214 filter(
215 lambda x: x.name in selected_columns,
216 modelfields + tuple(annotationfields),
217 )
218 )
219 kwargs["extra_columns"] = [
220 (field.name, library.column_for_field(field, accessor=field.name))
221 for field in extra_fields
222 if field.name not in self.get_table_class().base_columns
223 ]
225 return kwargs
227 def get_filterset_class(self):
228 filterset_modules = module_paths(
229 self.model, path="filtersets", suffix="FilterSet"
230 )
231 filterset_class = first_member_match(filterset_modules, GenericFilterSet)
232 return filterset_factory(self.model, filterset_class)
234 def _get_columns_choices(self, columns_exclude):
235 # lets start with the custom table fields
236 choices = {
237 key.name: capfirst(str(key) or key.name or "Nameless column")
238 for key in self.get_table().columns
239 }
240 # then add the model fields, but only the ones
241 # that are not automatically created (parent keys)
242 # and not the m2m relations and not any that are
243 # already part of the choices
244 choices |= {
245 field.name: pretty_name(getattr(field, "verbose_name", field.name))
246 for field in self.model._meta.get_fields()
247 if not getattr(field, "auto_created", False)
248 and not isinstance(field, ManyToManyRel)
249 and field.name not in choices.keys()
250 }
251 # finally we add any annotated fields
252 choices |= {key: key for key in self.get_queryset().query.annotations.keys()}
253 # now we drop all the choices that are listed in columns_exclude
254 choices = {
255 key: value for key, value in choices.items() if key not in columns_exclude
256 }
257 return choices.items()
259 def get_filterset_kwargs(self, filterset_class):
260 kwargs = super().get_filterset_kwargs(filterset_class)
261 kwargs["prefix"] = "filterset"
262 return kwargs
264 def get_filterset(self, filterset_class):
265 """
266 We override the `get_filterset` method, so we can add a
267 css class to the the selected filters
268 """
269 filterset = super().get_filterset(filterset_class)
271 # If the filterset form contains form data
272 # we add a CSS class to the element wrapping
273 # that field in HTML. This CSS class can be
274 # used to emphasize the fields that are used.
275 # To be able to compare the fields with the form
276 # data, we create a temporary mapping between
277 # widget_names and fields
278 fields = {}
279 for name, field in filterset.form.fields.items():
280 fields[name] = name
281 if hasattr(field.widget, "widgets_names"):
282 for widget_name in field.widget.widgets_names:
283 fields[name + widget_name] = name
284 if filterset.form.is_valid():
285 data = filterset.form.cleaned_data
286 for param in [param for param, value in data.items() if value]:
287 if fieldname := fields.get(param, None):
288 filterset.form.helper[fieldname].wrap(
289 Field, wrapper_class="filter-input-selected"
290 )
292 return filterset
294 def get_queryset(self):
295 queryset_methods = module_paths(
296 self.model, path="querysets", suffix="ListViewQueryset"
297 )
298 queryset = first_member_match(queryset_methods) or (lambda x: x)
299 return queryset(self.model.objects.all())
301 def get_table_pagination(self, table):
302 """
303 Override `get_table_pagination` from the tables2 TableMixinBase,
304 so we can set the table_pagination value as attribute of the table.
305 """
306 self.table_pagination = getattr(table, "table_pagination", None)
307 return super().get_table_pagination(table)
309 def get_context_data(self, *args, **kwargs):
310 context = super().get_context_data(*args, **kwargs)
311 table = context.get("table", None)
312 filterset = context.get("filter", None)
313 context["filterset_remember"] = (
314 self.request.GET.get("filterset-remember", "") == "on"
315 )
316 if table and filterset:
317 columns_exclude = filterset.form.columns_exclude
318 initial_columns = [
319 col.name for col in table.columns if col.name not in columns_exclude
320 ]
321 data = (
322 self.request.GET
323 if any(["choices" in key for key in self.request.GET.keys()])
324 else None
325 )
326 if choices := self._get_columns_choices(columns_exclude=columns_exclude):
327 context["columns_selector"] = ColumnsSelectorForm(
328 choices=choices,
329 initial={"choices": initial_columns},
330 prefix="choices",
331 data=data,
332 )
333 return context
336class Detail(GenericModelMixin, GenericModelPermissionRequiredMixin, DetailView):
337 """
338 Detail view for a generic model.
339 Access requires the `<model>_view` permission.
340 """
342 permission_action_required = "view"
345class Create(
346 GenericModelMixin,
347 GenericModelPermissionRequiredMixin,
348 SuccessMessageMixin,
349 CreateView,
350):
351 """
352 Create view for a generic model.
353 Access requires the `<model>_add` permission.
354 The form class is overridden by the first match from
355 the `first_member_match` helper.
356 """
358 template_name_suffix = "_create"
359 permission_action_required = "add"
361 def get_form_class(self):
362 form_modules = module_paths(self.model, path="forms", suffix="Form")
363 form_class = first_member_match(form_modules, GenericModelForm)
364 return modelform_factory(self.model, form_class)
366 def get_success_message(self, cleaned_data):
367 message_templates = template_names_via_mro(
368 self.model, suffix="_create_success_message.html"
369 )
370 template = select_template(message_templates)
371 return template.render({"object": self.object})
373 def get_success_url(self):
374 return self.object.get_create_success_url()
377class Delete(GenericModelMixin, GenericModelPermissionRequiredMixin, DeleteView):
378 """
379 Delete view for a generic model.
380 Access requires the `<model>_delete` permission.
381 """
383 permission_action_required = "delete"
385 def get_success_url(self):
386 if redirect := self.request.GET.get("redirect"):
387 return redirect
388 return reverse(
389 "apis_core:generic:list",
390 args=[self.request.resolver_match.kwargs["contenttype"]],
391 )
394class Update(
395 GenericModelMixin,
396 GenericModelPermissionRequiredMixin,
397 SuccessMessageMixin,
398 UpdateView,
399):
400 """
401 Update view for a generic model.
402 Access requires the `<model>_change` permission.
403 The form class is overridden by the first match from
404 the `first_member_match` helper.
405 """
407 permission_action_required = "change"
409 def get_form_class(self):
410 form_modules = module_paths(self.model, path="forms", suffix="Form")
411 form_class = first_member_match(form_modules, GenericModelForm)
412 return modelform_factory(self.model, form_class)
414 def get_success_message(self, cleaned_data):
415 message_templates = template_names_via_mro(
416 self.model, suffix="_update_success_message.html"
417 )
418 template = select_template(message_templates)
419 return template.render({"object": self.object})
421 def get_success_url(self):
422 return self.object.get_update_success_url()
425class Duplicate(GenericModelMixin, GenericModelPermissionRequiredMixin, View):
426 permission_action_required = "add"
428 def get(self, request, *args, **kwargs):
429 source_obj = get_object_or_404(self.model, pk=kwargs["pk"])
430 newobj = source_obj.duplicate()
432 message_templates = template_names_via_mro(
433 self.model, suffix="_duplicate_success_message.html"
434 )
435 template = select_template(message_templates)
436 messages.success(request, template.render({"object": source_obj}))
437 return redirect(newobj.get_edit_url())
440class Autocomplete(
441 GenericModelMixin,
442 GenericModelPermissionRequiredMixin,
443 autocomplete.Select2QuerySetView,
444):
445 """
446 Autocomplete view for a generic model.
447 Access requires the `<model>_view` permission.
448 The queryset is overridden by the first match from
449 the `first_member_match` helper.
450 """
452 permission_action_required = "view"
453 template_name_suffix = "_autocomplete_result"
455 def setup(self, *args, **kwargs):
456 super().setup(*args, **kwargs)
457 # We use a URI parameter to enable the create functionality in the
458 # autocomplete dropdown. It is not important what the value of the
459 # `create_field` is, because we override create_object anyway.
460 self.create_field = self.request.GET.get("create", None)
461 try:
462 template = select_template(self.get_template_names())
463 self.template = template.template.name
464 except TemplateDoesNotExist:
465 self.template = None
467 def get_queryset(self):
468 queryset_methods = module_paths(
469 self.model, path="querysets", suffix="AutocompleteQueryset"
470 )
471 queryset = first_member_match(queryset_methods)
472 if queryset:
473 return queryset(self.model, self.q)
474 return self.model.objects.filter(generate_search_filter(self.model, self.q))
476 def get_results(self, context):
477 external_only = self.kwargs.get("external_only", False)
478 results = [] if external_only else super().get_results(context)
479 queryset_methods = module_paths(
480 self.model, path="querysets", suffix="ExternalAutocomplete"
481 )
482 ExternalAutocomplete = first_member_match(queryset_methods)
483 if ExternalAutocomplete:
484 results.extend(ExternalAutocomplete().get_results(self.q))
485 return results
487 def create_object(self, value):
488 """
489 We try multiple approaches to create a model instance from a value:
490 * we first test if the value is an URL and if so we expect it to be
491 something that can be imported using one of the configured importers
492 and so we pass the value to the import logic.
493 * if the value is not a string, we try to pass it to the `create_from_string`
494 method of the model, if that does exist. Its the models responsibility to
495 implement this method and the method should somehow know how to create
496 model instance from the value...
497 * finally we pass the value to the `create_object` method from the DAL
498 view, which tries to pass it to `get_or_create` which likely also fails,
499 but this is expected and we raise a more useful exception.
500 """
501 try:
502 URLValidator()(value)
503 return self.queryset.model.import_from(value)
504 except ValidationError:
505 pass
506 try:
507 return self.queryset.model.create_from_string(value)
508 except AttributeError:
509 raise ImproperlyConfigured(
510 f'Model "{self.queryset.model._meta.verbose_name}" not configured to create from string'
511 )
513 def post(self, request, *args, **kwargs):
514 try:
515 with transaction.atomic():
516 return super().post(request, *args, **kwargs)
517 except Exception as e:
518 logger.debug(traceback.format_exc())
519 return http.JsonResponse({"error": str(e)})
522class Import(GenericModelMixin, GenericModelPermissionRequiredMixin, FormView):
523 template_name_suffix = "_import"
524 permission_action_required = "add"
526 def get_form_class(self):
527 form_modules = module_paths(self.model, path="forms", suffix="ImportForm")
528 form_class = first_member_match(form_modules, GenericImportForm)
529 return modelform_factory(self.model, form_class)
531 def form_valid(self, form):
532 self.object = form.cleaned_data["url"]
533 for field, error in getattr(self.object, "_import_errors", {}).items():
534 messages.error(self.request, f"Could not set {field}: {error}")
535 return super().form_valid(form)
537 def get_success_url(self):
538 return self.object.get_absolute_url()
541class SelectMergeOrEnrich(
542 GenericModelMixin, GenericModelPermissionRequiredMixin, FormView
543):
544 """
545 This view provides a simple form that allows to select other entities (also from
546 external sources, if set up) and on form submit redirects to the Enrich view.
547 """
549 template_name_suffix = "_selectmergeorenrich"
550 permission_action_required = "add"
551 form_class = GenericSelectMergeOrEnrichForm
553 def get_object(self, *args, **kwargs):
554 return get_object_or_404(self.model, pk=self.kwargs.get("pk"))
556 def get_context_data(self, *args, **kwargs):
557 context = super().get_context_data(*args, **kwargs)
558 context["object"] = self.get_object()
559 return context
561 def get_form_kwargs(self, *args, **kwargs):
562 kwargs = super().get_form_kwargs(*args, **kwargs)
563 kwargs["content_type"] = ContentType.objects.get_for_model(self.model)
564 return kwargs
566 def form_valid(self, form):
567 uri = form.cleaned_data["uri"]
568 if uri.isdigit():
569 return redirect(self.get_object().get_merge_url(uri))
570 return redirect(self.get_object().get_enrich_url() + f"?uri={uri}")
573class MergeWith(GenericModelMixin, GenericModelPermissionRequiredMixin, FormView):
574 """
575 Generic merge view.
576 """
578 permission_action_required = "change"
579 form_class = GenericMergeWithForm
580 template_name_suffix = "_merge"
582 def setup(self, *args, **kwargs):
583 super().setup(*args, **kwargs)
584 self.object = get_object_or_404(self.model, pk=self.kwargs["pk"])
585 self.other = get_object_or_404(self.model, pk=self.kwargs["otherpk"])
587 def get_context_data(self, **kwargs):
588 """
589 The context consists of the two objects that are merged as well
590 as a list of changes. Those changes are presented in the view as
591 a table with diffs
592 """
593 Change = namedtuple("Change", "field old new")
594 ctx = super().get_context_data(**kwargs)
595 ctx["changes"] = []
596 for field in self.object._meta.fields:
597 newval = self.object.get_field_value_after_merge(self.other, field)
598 ctx["changes"].append(
599 Change(field.verbose_name, getattr(self.object, field.name), newval)
600 )
601 ctx["object"] = self.object
602 ctx["other"] = self.other
603 return ctx
605 def form_valid(self, form):
606 self.object.merge_with([self.other])
607 messages.info(self.request, f"Merged values of {self.other} into {self.object}")
608 return super().form_valid(form)
610 def get_success_url(self):
611 return self.object.get_absolute_url()
614class Enrich(GenericModelMixin, GenericModelPermissionRequiredMixin, FormView):
615 """
616 Enrich an entity with data from an external source
617 Provides the user with a form to select the fields that should be updated.
618 """
620 permission_action_required = "change"
621 template_name_suffix = "_enrich"
622 form_class = GenericEnrichForm
623 importer_class = None
625 def setup(self, *args, **kwargs):
626 super().setup(*args, **kwargs)
627 self.object = get_object_or_404(self.model, pk=self.kwargs["pk"])
628 _, self.uri = get_autocomplete_data_and_normalized_uri(
629 self.request.GET.get("uri")
630 )
631 if not self.uri:
632 messages.error(self.request, "No uri parameter specified.")
634 def get(self, *args, **kwargs):
635 try:
636 uriobj = Uri.objects.get(uri=self.uri)
637 if uriobj.object_id != self.object.id:
638 messages.info(
639 self.request,
640 f"Object with URI {self.uri} already exists, you were redirected to the merge form.",
641 )
642 return redirect(self.object.get_merge_url(uriobj.object_id))
643 except Uri.DoesNotExist:
644 pass
645 return super().get(*args, **kwargs)
647 def get_context_data(self, **kwargs):
648 ctx = super().get_context_data(**kwargs)
649 ctx["object"] = self.object
650 ctx["uri"] = self.uri
651 return ctx
653 def get_form_kwargs(self, *args, **kwargs):
654 kwargs = super().get_form_kwargs(*args, **kwargs)
655 kwargs["instance"] = self.object
656 try:
657 self.data = self.model.fetch_from(self.request.GET.get("uri"))
658 kwargs["data"] = self.data
659 except ImproperlyConfigured as e:
660 messages.error(self.request, e)
661 return kwargs
663 def form_valid(self, form):
664 """
665 Go through all the form fields and extract the ones that
666 start with `update_` and that are set (those are the checkboxes that
667 select which fields to update).
668 Create a dict from those values, add the uri and pass the dict on to
669 the models `import_data` method.
670 """
671 data = {}
672 for key, values in self.request.POST.items():
673 if key.startswith("update_"):
674 key = key.removeprefix("update_")
675 data[key] = self.data[key]
676 if data:
677 self.object.import_data(data)
678 for field, error in getattr(self.object, "_import_errors", {}).items():
679 messages.error(self.request, f"Could not update {field}: {error}")
680 messages.info(self.request, f"Updated fields {data.keys()}")
681 return super().form_valid(form)
683 def get_success_url(self):
684 return self.object.get_absolute_url()