Merge pull request #273 from unicef/feature/283921

283921: Enhance Mapping Importer functionality and admin interface

Author: arsen-vs

src/country_workspace/admin/mapping_importer.py

@@ -10,12 +10,14 @@
 @admin.register(MappingImporter)
 class MappingImporterAdmin(BaseModelAdmin):
     readonly_fields = ("created_at", "last_modified", "created_by")
-    list_display = ("name", "data_checker", "created_by", "created_at", "last_modified")
+    list_display = ("name", "office", "data_checker", "created_by", "created_at", "last_modified")
     list_filter = (
-        "data_checker",
+        ("office", AutoCompleteFilter),
+        ("data_checker", AutoCompleteFilter),
         ("created_by", AutoCompleteFilter),
     )
-    search_fields = ("name",)
+    search_fields = ("name", "description")
+    autocomplete_fields = ("office", "data_checker")
 
     def save_model(self, request: HttpRequest, obj: MappingImporter, form: ModelForm, change: bool) -> None:
         if not change:

src/country_workspace/contrib/aurora/forms.py

@@ -14,6 +14,8 @@ class ImportAuroraForm(BaseImportForm):
     )
 
     def __init__(self, *args: Any, program: Program | None = None, **kwargs: Any) -> None:
+        if program:
+            kwargs["program"] = program
         super().__init__(*args, **kwargs)
         if program:
             self.fields["registration"].queryset = (

src/country_workspace/contrib/aurora/import_processing.py

@@ -1,4 +1,4 @@
-from typing import Any, NamedTuple
+from typing import Any, NamedTuple, NotRequired
 from itertools import chain
 from collections.abc import Mapping
 from functools import partial
@@ -18,6 +18,8 @@
 class Config(BatchNameConfig, ValidateModeConfig):
     registration_reference_pk: str | None
     master_detail: bool
+    household_mapping_id: NotRequired[int | None]
+    individual_mapping_id: NotRequired[int | None]
 
 
 class ImportResult(NamedTuple):

src/country_workspace/contrib/kobo/forms.py

@@ -3,6 +3,7 @@
 from django import forms
 
 from country_workspace.contrib.kobo.sync import make_client
+from country_workspace.models import Program
 from country_workspace.workspaces.admin.forms import BaseImportForm
 
 
@@ -14,7 +15,15 @@ class ImportKoboForm(BaseImportForm):
         help_text="Which field contains individual records",
     )
 
-    def __init__(self, *args: Any, kobo_country_code: str | None, **kwargs: Any) -> None:
+    def __init__(
+        self,
+        *args: Any,
+        kobo_country_code: str | None = None,
+        program: Program | None = None,
+        **kwargs: Any,
+    ) -> None:
+        kwargs["program"] = program
+
         super().__init__(*args, **kwargs)
         if kobo_country_code:
             client = make_client(kobo_country_code)

src/country_workspace/contrib/kobo/sync.py

@@ -1,7 +1,7 @@
 import re
 from collections.abc import Callable, Iterable
 from functools import partial
-from typing import Any, Final, TypedDict, cast, TYPE_CHECKING
+from typing import Any, Final, NotRequired, TypedDict, cast, TYPE_CHECKING
 from constance import config as constance_config
 from django.utils import timezone
 from requests import Session
@@ -11,8 +11,6 @@
 
 from country_workspace.contrib.kobo.exceptions import AlienFieldsError
 
-if TYPE_CHECKING:
-    from hope_flex_fields.models import DataChecker
 
 from country_workspace.contrib.kobo.api.client.auth import Auth
 from country_workspace.contrib.kobo.api.client.main import Client
@@ -26,10 +24,15 @@
 from country_workspace.utils.sync_log import get_kobo_sync_log_name
 from country_workspace.workspaces.admin.cleaners.validate import create_validation_jobs
 
+if TYPE_CHECKING:
+    from hope_flex_fields.models import DataChecker
+
 
 class Config(BatchNameConfig, ValidateModeConfig):
     project_id: str
     individual_records_field: str
+    household_mapping_id: NotRequired[int | None]
+    individual_mapping_id: NotRequired[int | None]
 
 
 ACCEPT_JSON_HEADERS: Final[dict[str, str]] = {"Accept": "application/json"}
@@ -93,11 +96,12 @@ def get_fullname_key(individual: Iterable[str]) -> str | None:
 
 def create_individuals(batch: Batch, household: Household, submission: Submission, config: Config) -> list[Individual]:
     individuals = []
+    individual_mapping_id = config.get("individual_mapping_id")
     for raw_individual in submission.get(config["individual_records_field"], []):
         individual_fields = preprocess(
             raw_individual,
             INDIVIDUAL_FIELDS_TO_UPPERCASE + TO_UPPERCASE_FIELDS,
-            partial(batch.program.apply_mapping_importer, Individual),
+            partial(batch.program.apply_mapping_importer, Individual, mapping_id=individual_mapping_id),
             partial(batch.program.apply_default_fields, Individual),
         )
         fullname = get_fullname_key(individual_fields)
@@ -117,11 +121,12 @@ def create_individuals(batch: Batch, household: Household, submission: Submissio
 def create_household(
     batch: Batch, submission: Submission, config: Config, id_generator: Callable[[], int]
 ) -> Household:
+    household_mapping_id = config.get("household_mapping_id")
     raw_household_fields = extract_household_data(submission, config["individual_records_field"])
     household_fields = preprocess(
         raw_household_fields,
         HOUSEHOLD_FIELDS_TO_UPPERCASE,
-        partial(batch.program.apply_mapping_importer, Household),
+        partial(batch.program.apply_mapping_importer, Household, mapping_id=household_mapping_id),
         partial(batch.program.apply_default_fields, Household),
     )
     household_fields["household_id"] = id_generator()

src/country_workspace/datasources/rdi/config.py

@@ -33,6 +33,8 @@ class Config(BatchNameConfig, ValidateModeConfig):
     household_label: NotRequired[str]
     people_prefix: NotRequired[str]
     first_line: int
+    household_mapping_id: NotRequired[int | None]
+    individual_mapping_id: NotRequired[int | None]
 
 
 class SheetName(StrEnum):

src/country_workspace/datasources/rdi/processors.py

@@ -109,15 +109,17 @@ def read_sheets(config: Config, filepath: str, *sheet_names: str) -> Generator[S
 
 def process_households(sheet: Sheet, job: AsyncJob, batch: Batch, config: Config) -> Mapping[int, Household]:
     mapping = {}
+    household_mapping_id = config.get("household_mapping_id")
     transform_row = compose(
         clean_field_names,
-        partial(job.program.apply_mapping_importer, Household),
+        partial(job.program.apply_mapping_importer, Household, mapping_id=household_mapping_id),
         partial(job.program.apply_default_fields, Household),
     )
 
     for row in sheet:
         if (household_key := get_value(row, config["household_id_column"])) in mapping:
             raise SheetProcessingError(SheetName.HOUSEHOLDS, household_key)
+
         try:
             mapping[household_key] = cast(
                 "Household",
@@ -141,9 +143,10 @@ def process_beneficiaries(
     people_prefix = config.get("people_prefix") if household_mapping is None else None
     household_id_column = config.get("household_id_column") if household_mapping is not None else None
     sheet_name = SheetName.PEOPLE if household_mapping is None else SheetName.INDIVIDUALS
+    individual_mapping_id = config.get("individual_mapping_id")
     transform_row = compose(
         clean_field_names,
-        partial(job.program.apply_mapping_importer, Individual),
+        partial(job.program.apply_mapping_importer, Individual, mapping_id=individual_mapping_id),
         partial(job.program.apply_default_fields, Individual),
     )
 
@@ -155,6 +158,7 @@ def process_beneficiaries(
         cleaned_row, name_column = normalize_row_structure(row, people_prefix)
         name = cleaned_row.get(name_column) if name_column else ""
         household = get_hh_for_ind(cleaned_row, household_id_column, household_mapping)
+
         try:
             mapping[beneficiary_key] = cast(
                 "Individual",
@@ -235,7 +239,7 @@ def _sync_ind_pks(households_mapping: dict, individuals_mapping: dict) -> None:
     pk_mapping = {v.flex_fields.get("individual_id"): v.pk for _, v in individuals_mapping.items()}
 
     for v in households_mapping.values():
-        hh_flex_fields = v.flex_fields
+        hh_flex_fields = v.flex_fields.copy()
         hh_flex_fields["head_of_household"] = pk_mapping.get(v.flex_fields.get("head_of_household"))
         hh_flex_fields["primary_collector"] = pk_mapping.get(v.flex_fields.get("primary_collector"))
         if alt_id := v.flex_fields.get("alternate_collector"):  # is optional

src/country_workspace/migrations/0033_update_mapping_importer_with_office.py

@@ -0,0 +1,39 @@
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("country_workspace", "0032_program_system_fields"),
+        ("hope_flex_fields", "0013_fielddefinition_validated_alter_datachecker_id_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="mappingimporter",
+            name="office",
+            field=models.ForeignKey(
+                null=True,
+                blank=True,
+                on_delete=django.db.models.deletion.CASCADE,
+                related_name="mapping_importers",
+                to="country_workspace.office",
+                help_text="Business Area (Office) this mapping belongs to",
+            ),
+        ),
+        # Change data_checker from OneToOneField to ForeignKey
+        migrations.AlterField(
+            model_name="mappingimporter",
+            name="data_checker",
+            field=models.ForeignKey(
+                on_delete=django.db.models.deletion.CASCADE,
+                related_name="mapping_importers",
+                to="hope_flex_fields.datachecker",
+                help_text="DataChecker (Household/Individual) this mapping is valid for",
+            ),
+        ),
+        migrations.AlterUniqueTogether(
+            name="mappingimporter",
+            unique_together={("office", "name")},
+        ),
+    ]

src/country_workspace/models/mapping_importer.py

@@ -9,9 +9,20 @@
 
 
 class MappingImporter(BaseModel):
-    data_checker = models.OneToOneField(DataChecker, on_delete=models.CASCADE, related_name="%(class)s")
     name = models.CharField(max_length=255)
     description = models.CharField(max_length=255, blank=True)
+    office = models.ForeignKey(
+        "Office",
+        on_delete=models.CASCADE,
+        related_name="mapping_importers",
+        help_text=_("Business Area (Office) this mapping belongs to"),
+    )
+    data_checker = models.ForeignKey(
+        DataChecker,
+        on_delete=models.CASCADE,
+        related_name="mapping_importers",
+        help_text=_("DataChecker (Household/Individual) this mapping is valid for"),
+    )
     rules = models.TextField(
         blank=True,
         default="",
@@ -27,6 +38,7 @@ class MappingImporter(BaseModel):
     class Meta:
         verbose_name = _("Mapping Importer")
         verbose_name_plural = _("Mapping Importers")
+        unique_together = [["office", "name"]]
 
     def __str__(self) -> str:
         return self.name

src/country_workspace/models/program.py

@@ -1,10 +1,9 @@
-from contextlib import suppress
 from typing import TYPE_CHECKING, Any
 from collections.abc import Iterable
 from enum import StrEnum
 
-from django.core.exceptions import ObjectDoesNotExist
 from django.db import models
+from django.db.models import Q
 from django.utils.translation import gettext as _
 from hope_flex_fields.models import DataChecker
 from strategy_field.fields import StrategyField
@@ -161,13 +160,28 @@ def serialize(self, data: list[dict]) -> Iterable:
         return data
 
     def apply_mapping_importer(
-        self, m: type[Validable] | Validable, data: dict[str, str | int | bool]
+        self,
+        m: type[Validable] | Validable,
+        data: dict[str, str | int | bool],
+        mapping_id: int | None = None,
     ) -> dict[str, str | int | bool]:
         """Apply mapping importer from the checker's mappingimporter, if any."""
-        if (checker := self.get_checker_for(m)) is None:
-            return data
-        with suppress(ObjectDoesNotExist):
-            checker.mappingimporter.apply(data)
+        from country_workspace.models import MappingImporter
+
+        mapping_importers = []
+        if mapping_id:
+            mapping_importer = MappingImporter.objects.filter(id=mapping_id).first()
+            if mapping_importer:
+                mapping_importers = [mapping_importer]
+
+        elif (checker := self.get_checker_for(m)) is not None:
+            mapping_importers = list(
+                checker.mapping_importers.filter(Q(office=self.country_office) | Q(office__isnull=True))
+            )
+
+        for mapping_importer in mapping_importers:
+            mapping_importer.apply(data)
+
         return data
 
     def get_default_fields_for(self, m: type[Validable] | Validable) -> dict[str, Any]: