TP2000-1387 Async Bulk Edit - Python 3.12

measures/editors.py

@@ -0,0 +1,118 @@
+from typing import Dict
+from typing import List
+from typing import Type
+
+from django.db.transaction import atomic
+
+from workbaskets import models as workbasket_models
+from measures import models as measure_models
+from common.util import TaricDateRange
+from common.validators import UpdateType
+from common.models.utils import override_current_transaction
+from measures.util import update_measure_components
+from measures.util import update_measure_condition_components
+from measures.util import update_measure_excluded_geographical_areas
+from measures.util import update_measure_footnote_associations
+
+
+class MeasuresEditor:
+    """Utility class used to edit measures from measures wizard accumulated
+    data."""
+
+    workbasket: Type["workbasket_models.WorkBasket"]
+    """The workbasket with which created measures will be associated."""
+
+    selected_measures: List
+    """ The measures in which the edits will apply to."""
+
+    data: Dict
+    """Validated, cleaned and accumulated data created by the Form instances of
+    `MeasureEditWizard`."""
+
+    def __init__(
+        self,
+        workbasket: Type["workbasket_models.WorkBasket"],
+        selected_measures: List,
+        data: Dict,
+    ):
+        self.workbasket = workbasket
+        self.selected_measures = selected_measures
+        self.data = data
+
+    @atomic
+    def edit_measures(self) -> List["measure_models.Measure"]:
+        """
+        Returns a list of the edited measures.
+
+        `data` must be a dictionary
+        of the accumulated cleaned / validated data created from the
+        `MeasureEditWizard`.
+        """
+
+        with override_current_transaction(
+            transaction=self.workbasket.current_transaction,
+        ):
+            new_start_date = self.data.get("start_date", None)
+            new_end_date = self.data.get("end_date", False)
+            new_quota_order_number = self.data.get("order_number", None)
+            new_generating_regulation = self.data.get("generating_regulation", None)
+            new_duties = self.data.get("duties", None)
+            new_exclusions = [
+                e["excluded_area"]
+                for e in self.data.get("formset-geographical_area_exclusions", [])
+            ]
+
+            edited_measures = []
+
+            if self.selected_measures:
+                for measure in self.selected_measures:
+                    new_measure = measure.new_version(
+                        workbasket=self.workbasket,
+                        update_type=UpdateType.UPDATE,
+                        valid_between=TaricDateRange(
+                            lower=(
+                                new_start_date
+                                if new_start_date
+                                else measure.valid_between.lower
+                            ),
+                            upper=(
+                                new_end_date
+                                if new_end_date
+                                else measure.valid_between.upper
+                            ),
+                        ),
+                        order_number=(
+                            new_quota_order_number
+                            if new_quota_order_number
+                            else measure.order_number
+                        ),
+                        generating_regulation=(
+                            new_generating_regulation
+                            if new_generating_regulation
+                            else measure.generating_regulation
+                        ),
+                    )
+                    update_measure_components(
+                        measure=new_measure,
+                        duties=new_duties,
+                        workbasket=self.workbasket,
+                    )
+                    update_measure_condition_components(
+                        measure=new_measure,
+                        workbasket=self.workbasket,
+                    )
+                    update_measure_excluded_geographical_areas(
+                        edited="geographical_area_exclusions"
+                        in self.data.get("fields_to_edit", []),
+                        measure=new_measure,
+                        exclusions=new_exclusions,
+                        workbasket=self.workbasket,
+                    )
+                    update_measure_footnote_associations(
+                        measure=new_measure,
+                        workbasket=self.workbasket,
+                    )
+
+                    edited_measures.append(new_measure.id)
+
+            return edited_measures

measures/forms/wizard.py

@@ -781,7 +781,7 @@ def __init__(self, *args, **kwargs):
         )
 
 
-class MeasureStartDateForm(forms.Form):
+class MeasureStartDateForm(forms.Form, SerializableFormMixin):
     start_date = DateInputFieldFixed(
         label="Start date",
         help_text="For example, 27 3 2008",
@@ -819,8 +819,34 @@ def clean(self):
 
         return cleaned_data
 
+    @classmethod
+    def serializable_init_kwargs(cls, kwargs: Dict) -> Dict:
+        selected_measures = kwargs.get("selected_measures")
+        selected_measures_pks = []
+        for measure in selected_measures:
+            selected_measures_pks.append(measure.id)
+
+        serializable_kwargs = {
+            "selected_measures": selected_measures_pks,
+        }
+
+        return serializable_kwargs
+
+    @classmethod
+    def deserialize_init_kwargs(cls, form_kwargs: Dict) -> Dict:
+        serialized_selected_measures_pks = form_kwargs.get("selected_measures")
+        deserialized_selected_measures = models.Measure.objects.filter(
+            pk__in=serialized_selected_measures_pks
+        )
+
+        kwargs = {
+            "selected_measures": deserialized_selected_measures,
+        }
+
+        return kwargs
+
 
-class MeasureEndDateForm(forms.Form):
+class MeasureEndDateForm(forms.Form, SerializableFormMixin):
     end_date = DateInputFieldFixed(
         label="End date",
         help_text="For example, 27 3 2008",
@@ -861,8 +887,34 @@ def clean(self):
 
         return cleaned_data
 
+    @classmethod
+    def serializable_init_kwargs(cls, kwargs: Dict) -> Dict:
+        selected_measures = kwargs.get("selected_measures")
+        selected_measures_pks = []
+        for measure in selected_measures:
+            selected_measures_pks.append(measure.id)
+
+        serializable_kwargs = {
+            "selected_measures": selected_measures_pks,
+        }
+
+        return serializable_kwargs
+
+    @classmethod
+    def deserialize_init_kwargs(cls, form_kwargs: Dict) -> Dict:
+        serialized_selected_measures_pks = form_kwargs.get("selected_measures")
+        deserialized_selected_measures = models.Measure.objects.filter(
+            pk__in=serialized_selected_measures_pks
+        )
+
+        kwargs = {
+            "selected_measures": deserialized_selected_measures,
+        }
+
+        return kwargs
+
 
-class MeasureRegulationForm(forms.Form):
+class MeasureRegulationForm(forms.Form, SerializableFormMixin):
     generating_regulation = AutoCompleteField(
         label="Regulation ID",
         help_text="Select the regulation which provides the legal basis for the measures.",
@@ -888,8 +940,34 @@ def __init__(self, *args, **kwargs):
             ),
         )
 
+    @classmethod
+    def serializable_init_kwargs(cls, kwargs: Dict) -> Dict:
+        selected_measures = kwargs.get("selected_measures")
+        selected_measures_pks = []
+        for measure in selected_measures:
+            selected_measures_pks.append(measure.id)
+
+        serializable_kwargs = {
+            "selected_measures": selected_measures_pks,
+        }
+
+        return serializable_kwargs
+
+    @classmethod
+    def deserialize_init_kwargs(cls, form_kwargs: Dict) -> Dict:
+        serialized_selected_measures_pks = form_kwargs.get("selected_measures")
+        deserialized_selected_measures = models.Measure.objects.filter(
+            pk__in=serialized_selected_measures_pks
+        )
+
+        kwargs = {
+            "selected_measures": deserialized_selected_measures,
+        }
+
+        return kwargs
+
 
-class MeasureDutiesForm(forms.Form):
+class MeasureDutiesForm(forms.Form, SerializableFormMixin):
     duties = forms.CharField(
         label="Duties",
         help_text="Enter the duty that applies to the measures.",
@@ -932,6 +1010,32 @@ def clean(self):
 
         return cleaned_data
 
+    @classmethod
+    def serializable_init_kwargs(cls, kwargs: Dict) -> Dict:
+        selected_measures = kwargs.get("selected_measures")
+        selected_measures_pks = []
+        for measure in selected_measures:
+            selected_measures_pks.append(measure.id)
+
+        serializable_kwargs = {
+            "selected_measures": selected_measures_pks,
+        }
+
+        return serializable_kwargs
+
+    @classmethod
+    def deserialize_init_kwargs(cls, form_kwargs: Dict) -> Dict:
+        serialized_selected_measures_pks = form_kwargs.get("selected_measures")
+        deserialized_selected_measures = models.Measure.objects.filter(
+            pk__in=serialized_selected_measures_pks
+        )
+
+        kwargs = {
+            "selected_measures": deserialized_selected_measures,
+        }
+
+        return kwargs
+
 
 class MeasureGeographicalAreaExclusionsForm(forms.Form):
     excluded_area = forms.ModelChoiceField(
@@ -965,7 +1069,7 @@ def __init__(self, *args, **kwargs):
         )
 
 
-class MeasureGeographicalAreaExclusionsFormSet(FormSet):
+class MeasureGeographicalAreaExclusionsFormSet(FormSet, SerializableFormMixin):
     """Allows editing the geographical area exclusions of multiple measures in
     `MeasureEditWizard`."""
 

measures/migrations/0017_measuresbulkeditor.py

@@ -0,0 +1,84 @@
+# Generated by Django 4.2.15 on 2024-09-02 16:06
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+import django_fsm
+import measures.models.bulk_processing
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ("workbaskets", "0008_datarow_dataupload"),
+        ("measures", "0016_measuresbulkcreator"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="MeasuresBulkEditor",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+                (
+                    "task_id",
+                    models.CharField(blank=True, max_length=50, null=True, unique=True),
+                ),
+                (
+                    "processing_state",
+                    django_fsm.FSMField(
+                        choices=[
+                            ("AWAITING_PROCESSING", "Awaiting processing"),
+                            ("CURRENTLY_PROCESSING", "Currently processing"),
+                            ("SUCCESSFULLY_PROCESSED", "Successfully processed"),
+                            ("FAILED_PROCESSING", "Failed processing"),
+                            ("CANCELLED", "Cancelled"),
+                        ],
+                        db_index=True,
+                        default="AWAITING_PROCESSING",
+                        editable=False,
+                        max_length=50,
+                        protected=True,
+                    ),
+                ),
+                (
+                    "successfully_processed_count",
+                    models.PositiveIntegerField(default=0),
+                ),
+                ("form_data", models.JSONField()),
+                ("form_kwargs", models.JSONField()),
+                ("selected_measures", models.JSONField()),
+                (
+                    "user",
+                    models.ForeignKey(
+                        editable=False,
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+                (
+                    "workbasket",
+                    models.ForeignKey(
+                        editable=False,
+                        null=True,
+                        on_delete=measures.models.bulk_processing.REVOKE_TASKS_AND_SET_NULL,
+                        to="workbaskets.workbasket",
+                    ),
+                ),
+            ],
+            options={
+                "abstract": False,
+            },
+        ),
+    ]

measures/models/__init__.py

@@ -1,5 +1,6 @@
 from measures.models.bulk_processing import BulkProcessor
 from measures.models.bulk_processing import MeasuresBulkCreator
+from measures.models.bulk_processing import MeasuresBulkEditor
 from measures.models.bulk_processing import ProcessingState
 from measures.models.tracked_models import AdditionalCodeTypeMeasureType
 from measures.models.tracked_models import DutyExpression
@@ -23,6 +24,7 @@
     # - Classes exported from bulk_processing.py.
     "BulkProcessor",
     "MeasuresBulkCreator",
+    "MeasuresBulkEditor",
     "ProcessingState",
     # - Classes exported from tracked_model.py.
     "AdditionalCodeTypeMeasureType",