add form validation runner classes

edc_data_manager/admin/__init__.py

@@ -8,3 +8,4 @@
 from .data_dictionary_admin import DataDictionaryAdmin
 from .data_query_admin import DataQueryAdmin
 from .query_rule_admin import QueryRuleAdmin
+from .validation_errors_admin import ValidationErrorsAdmin

edc_data_manager/admin/actions.py

@@ -1,40 +1,65 @@
 from uuid import uuid4
 
 from django.conf import settings
-from django.contrib import messages
+from django.contrib import admin, messages
 from django.core.exceptions import ObjectDoesNotExist
 from django.urls.base import reverse
 from django.utils.html import format_html
 from django.utils.safestring import mark_safe
-from edc_constants.constants import CLOSED, OPEN
+from edc_appointment.constants import IN_PROGRESS_APPT
+from edc_constants.constants import CLOSED, DONE, NEW, OPEN
 from edc_utils import formatted_datetime, get_utcnow
 
+from ..form_validation_runners import SingleFormValidationRunner
 from ..models import QueryRule
 from ..rule import update_query_rules
 
 DATA_MANAGER_ENABLED = getattr(settings, "DATA_MANAGER_ENABLED", True)
 
 
-def toggle_active_flag(modeladmin, request, queryset):
+@admin.action(description="Refresh selected")
+def validation_error_refresh(modeladmin, request, queryset):
     for obj in queryset:
-        obj.active = False if obj.active else True
+        runner = SingleFormValidationRunner(validation_error_obj=obj)
+        runner.run()
+
+
+@admin.action(description="Mark selected as done")
+def validation_error_flag_as_done(modeladmin, request, queryset):
+    for obj in queryset:
+        obj.status = DONE
         obj.save()
 
 
-toggle_active_flag.short_description = (
-    f"Toggle Active/Inactive {QueryRule._meta.verbose_name_plural}"
-)
+@admin.action(description="Mark selected as in progress")
+def validation_error_flag_as_in_progress(modeladmin, request, queryset):
+    for obj in queryset:
+        obj.status = IN_PROGRESS_APPT
+        obj.save()
 
 
-def toggle_dm_status(modeladmin, request, queryset):
+@admin.action(description="Mark selected as new")
+def validation_error_flag_as_new(modeladmin, request, queryset):
     for obj in queryset:
-        obj.status = OPEN if obj.status != OPEN else CLOSED
+        obj.status = NEW
         obj.save()
 
 
-toggle_dm_status.short_description = "Toggle DM Status (OPEN/CLOSED)"
+@admin.action(description=f"Toggle Active/Inactive {QueryRule._meta.verbose_name_plural}")
+def toggle_active_flag(modeladmin, request, queryset):
+    for obj in queryset:
+        obj.active = False if obj.active else True
+        obj.save()
+
 
+@admin.action(description="Toggle DM Status (OPEN/CLOSED)")
+def toggle_dm_status(modeladmin, request, queryset):
+    for obj in queryset:
+        obj.status = OPEN if obj.status != OPEN else CLOSED
+        obj.save()
 
+
+@admin.action(description=f"Copy {QueryRule._meta.verbose_name}")
 def copy_query_rule_action(modeladmin, request, queryset):
     if queryset.count() == 1:
         obj = queryset[0]
@@ -65,9 +90,7 @@ def copy_query_rule_action(modeladmin, request, queryset):
         )
 
 
-copy_query_rule_action.short_description = f"Copy {QueryRule._meta.verbose_name}"
-
-
+@admin.action(description=f"Run selected {QueryRule._meta.verbose_name_plural}")
 def update_query_rules_action(modeladmin, request, queryset):
     if not DATA_MANAGER_ENABLED:
         msg = (
@@ -97,8 +120,3 @@ def update_query_rules_action(modeladmin, request, queryset):
                 mark_safe(results.get("resolved")),  # nosec B703, B308
             )
         messages.add_message(request, messages.SUCCESS, msg)
-
-
-update_query_rules_action.short_description = (
-    f"Run selected {QueryRule._meta.verbose_name_plural}"
-)

edc_data_manager/admin/validation_errors_admin.py

@@ -0,0 +1,133 @@
+from textwrap import wrap
+
+from django.contrib import admin
+from django.utils.html import format_html
+from django_audit_fields import audit_fieldset_tuple
+from edc_model_admin.dashboard import ModelAdminSubjectDashboardMixin
+from edc_sites.admin import SiteModelAdminMixin
+
+from ..admin_site import edc_data_manager_admin
+from ..models import ValidationErrors
+from .actions import (
+    validation_error_flag_as_done,
+    validation_error_flag_as_in_progress,
+    validation_error_flag_as_new,
+    validation_error_refresh,
+)
+
+
+@admin.register(ValidationErrors, site=edc_data_manager_admin)
+class ValidationErrorsAdmin(
+    SiteModelAdminMixin,
+    ModelAdminSubjectDashboardMixin,
+    admin.ModelAdmin,
+):
+    list_per_page = 15
+    show_cancel = True
+    actions = [
+        validation_error_flag_as_done,
+        validation_error_flag_as_in_progress,
+        validation_error_flag_as_new,
+        validation_error_refresh,
+    ]
+
+    fieldsets = (
+        (
+            None,
+            {
+                "fields": (
+                    "verbose_name",
+                    "subject_identifier",
+                    "visit_code",
+                    "visit_code_sequence",
+                    "src_report_datetime",
+                    "src_revision",
+                    "site",
+                )
+            },
+        ),
+        (
+            "Message",
+            {
+                "fields": (
+                    "field_name",
+                    "label_lower",
+                    "message",
+                    "raw_message",
+                )
+            },
+        ),
+        (
+            "Session",
+            {"fields": ("session_id", "session_datetime")},
+        ),
+        (
+            "Status",
+            {"fields": ("status",)},
+        ),
+        audit_fieldset_tuple,
+    )
+
+    list_display = (
+        "subject_identifier",
+        "dashboard",
+        "document",
+        "error_msg",
+        "field_name",
+        "response",
+        "visit",
+        "status",
+    )
+
+    list_filter = (
+        "verbose_name",
+        "field_name",
+        "visit_code",
+        "visit_code_sequence",
+        "status",
+        "short_message",
+        "session_id",
+        "session_datetime",
+        "site",
+    )
+
+    readonly_fields = (
+        "session_id",
+        "session_datetime",
+        "verbose_name",
+        "raw_message",
+        "src_revision",
+        "src_report_datetime",
+        "src_modified_datetime",
+        "src_user_modified",
+        "subject_identifier",
+        "visit_code",
+        "visit_code_sequence",
+        "site",
+        "label_lower",
+        "field_name",
+        "message",
+    )
+
+    radio_fields = {
+        "status": admin.VERTICAL,
+    }
+
+    search_fields = (
+        "message",
+        "label_lower",
+        "field_name",
+        "subject_identifier",
+    )
+
+    @admin.display(description="Message", ordering="short_message")
+    def error_msg(self, obj):
+        return format_html("<BR>".join(wrap(obj.short_message, 45)).replace(" ", "&nbsp"))
+
+    @admin.display(description="Visit", ordering="visit_code")
+    def visit(self, obj):
+        return f"{obj.visit_code}.{obj.visit_code_sequence}"
+
+    @admin.display(description="Document", ordering="verbose_name")
+    def document(self, obj):
+        return obj.verbose_name

edc_data_manager/auth_objects.py

@@ -16,6 +16,8 @@
 data_manager = [
     "edc_data_manager.add_dataquery",
     "edc_data_manager.add_queryrule",
+    "edc_data_manager.view_validationerrors",
+    "edc_data_manager.change_validationerrors",
     "edc_data_manager.change_dataquery",
     "edc_data_manager.change_queryrule",
     "edc_data_manager.delete_dataquery",

edc_data_manager/form_validation_runners/__init__.py

@@ -0,0 +1,3 @@
+from .exceptions import FormValidationRunnerError
+from .form_validation_runner import FormValidationRunner
+from .single_form_validation_runner import SingleFormValidationRunner

edc_data_manager/form_validation_runners/exceptions.py

@@ -0,0 +1,2 @@
+class FormValidationRunnerError(Exception):
+    pass

edc_data_manager/form_validation_runners/form_validation_runner.py

@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+import html
+import uuid
+from typing import TYPE_CHECKING, Any, Type
+
+from bs4 import BeautifulSoup
+from django.db.models import ForeignKey, ManyToManyField, OneToOneField, QuerySet
+from edc_utils import get_utcnow
+from tqdm import tqdm
+
+from ..models import ValidationErrors
+
+if TYPE_CHECKING:
+    from django.forms import ModelForm
+
+
+class FormValidationRunner:
+    """Rerun modelform validation on all instances of a model.
+
+    Usage:
+        runner = FormValidationRunner(MyModelform)
+        runner.run()
+    """
+
+    def __init__(
+        self,
+        modelform_cls: Type[ModelForm] | None = None,
+        extra_formfields: list[str] | None = None,
+        ignore_formfields: list[str] | None = None,
+        verbose: bool | None = None,
+        filter_options: dict[str, str] | None = None,
+    ):
+        self.session_id = uuid.uuid4()
+        self.session_datetime = get_utcnow()
+        self.verbose = True if verbose is None else verbose
+        self.modelform_cls = modelform_cls
+        self.model_cls = modelform_cls._meta.model
+        self.extra_formfields = extra_formfields or []
+        self.ignore_formfields = ignore_formfields or []
+        self.filter_options = filter_options
+
+    def run(self):
+        self.delete_validation_errors()
+        total = self.queryset.count()
+        for model_obj in tqdm(self.queryset.order_by("created"), total=total):
+            data = self.get_form_data(model_obj)
+            form = self.modelform_cls(data, instance=model_obj)
+            form.is_valid()
+            errors = {k: v for k, v in form._errors.items() if k not in self.ignore_formfields}
+            if errors:
+                for k, v in errors.items():
+                    validation_error_obj = self.write_to_db(k, v, model_obj)
+                    if self.verbose:
+                        print(validation_error_obj)
+
+    def delete_validation_errors(self) -> None:
+        ValidationErrors.objects.filter(label_lower=self.model_cls._meta.label_lower).delete()
+
+    @property
+    def queryset(self) -> QuerySet:
+        opts = {}
+        if self.filter_options:
+            opts = dict(**self.filter_options)
+        return self.model_cls.objects.filter(**opts)
+
+    def get_form_data(self, model_obj) -> dict:
+        data = {
+            k: v
+            for k, v in model_obj.__dict__.items()
+            if not k.startswith("_") and not k.endswith("_id")
+        }
+        for fld_cls in model_obj._meta.get_fields():
+            if isinstance(fld_cls, (ForeignKey, OneToOneField)):
+                try:
+                    obj_fld_id = getattr(model_obj, fld_cls.name).id
+                except AttributeError:
+                    rel_obj = None
+                else:
+                    rel_obj = fld_cls.related_model.objects.get(id=obj_fld_id)
+                data.update({fld_cls.name: rel_obj})
+            elif isinstance(fld_cls, (ManyToManyField,)):
+                data.update({fld_cls.name: getattr(model_obj, fld_cls.name).all()})
+            else:
+                pass
+        try:
+            data.update(subject_visit=model_obj.subject_visit)
+        except AttributeError:
+            data.update(subject_identifier=model_obj.subject_identifier)
+        for extra_formfield in self.extra_formfields:
+            data.update({extra_formfield: getattr(model_obj, extra_formfield)})
+        return data
+
+    def write_to_db(self, k: str, v: Any, model_obj) -> ValidationErrors:
+        model_obj_or_related_visit = getattr(model_obj, "subject_visit", model_obj)
+        subject_identifier = model_obj_or_related_visit.subject_identifier
+        visit_code = model_obj_or_related_visit.visit_code
+        visit_code_sequence = model_obj_or_related_visit.visit_code_sequence
+        raw_message = html.unescape(v.as_text())
+        message = BeautifulSoup(raw_message, "html.parser").text
+        try:
+            response = getattr(model_obj, k)
+        except AttributeError:
+            response = None
+        return ValidationErrors.objects.create(
+            session_id=self.session_id,
+            session_datetime=self.session_datetime,
+            label_lower=model_obj._meta.label_lower,
+            verbose_name=model_obj._meta.verbose_name,
+            subject_identifier=subject_identifier,
+            visit_code=visit_code,
+            visit_code_sequence=visit_code_sequence,
+            field_name=k,
+            raw_message=raw_message,
+            message=message,
+            short_message=message[:250],
+            response=str(response),
+            src_id=model_obj.id,
+            src_revision=model_obj.revision,
+            src_report_datetime=getattr(model_obj, "report_datetime", None),
+            src_modified_datetime=model_obj.modified,
+            src_user_modified=model_obj.user_modified,
+            site=model_obj.site,
+            extra_formfields=",".join(self.extra_formfields),
+            ignore_formfields=",".join(self.ignore_formfields),
+        )