add ! import mapping rules, profiles

Author: vitali-yanushchyk-valor

pyproject.toml

@@ -46,6 +46,7 @@ dependencies = [
   "hope-flex-fields>=0.6.2",
   "hope-smart-export>=0.3",
   "hope-smart-import>=0.3",
+  "jmespath>=1.0.1",
   "openpyxl>=3.1.5",
   "pillow>=11.2.1",
   "psycopg2-binary>=2.9.9",

src/country_workspace/admin/__init__.py

@@ -12,6 +12,7 @@
 from .individual import IndividualAdmin
 from .job import AsyncJobAdmin
 from .locations import AreaAdmin, AreaTypeAdmin, CountryAdmin
+from .mapping import FieldMappingRuleAdmin, MappingProfileAdmin
 from .office import OfficeAdmin
 from .program import ProgramAdmin
 from .rdp import RdpAdmin
@@ -36,8 +37,10 @@
     "BeneficiaryGroupAdmin",
     "ConstanceAdmin",
     "CountryAdmin",
+    "FieldMappingRuleAdmin",
     "HouseholdAdmin",
     "IndividualAdmin",
+    "MappingProfileAdmin",
     "OfficeAdmin",
     "ProgramAdmin",
     "RdpAdmin",

src/country_workspace/admin/mapping.py

@@ -0,0 +1,58 @@
+from django.contrib import admin
+from django.db import models
+from django.http import HttpRequest
+
+from ..models import MappingProfile, FieldMappingRule
+from .base import BaseModelAdmin
+
+
+@admin.register(MappingProfile)
+class MappingProfileAdmin(BaseModelAdmin):
+    list_display = ("name", "source_type", "import_schema", "is_active", "inheritance")
+    list_filter = ("source_type", "import_schema", "program", "is_active")
+    search_fields = ("name", "description")
+    filter_horizontal = ("program",)
+    fields = (
+        "name",
+        "description",
+        "parent",
+        "source_type",
+        "import_schema",
+        "program",
+        "is_active",
+        "created_by",
+        "created_at",
+        "updated_at",
+    )
+    readonly_fields = ("created_by", "created_at", "updated_at")
+
+    @admin.display(description="Inheritance Chain")
+    def inheritance(self, obj: MappingProfile) -> str:
+        return obj.get_inheritance_chain()
+
+    def get_queryset(self, request: HttpRequest) -> models.QuerySet[MappingProfile]:
+        return super().get_queryset(request).select_related("parent", "created_by").prefetch_related("program")
+
+
+@admin.register(FieldMappingRule)
+class FieldMappingRuleAdmin(BaseModelAdmin):
+    list_display = ("name", "profile", "order", "is_active")
+    list_filter = ("profile", "is_active")
+    search_fields = ("name", "description", "profile__name")
+    ordering = ("profile", "order", "name")
+    fields = (
+        "name",
+        "description",
+        "profile",
+        "expression",
+        "order",
+        "is_active",
+        "created_by",
+        "created_at",
+        "updated_at",
+    )
+
+    readonly_fields = ("created_by", "created_at", "updated_at")
+
+    def get_queryset(self, request: HttpRequest) -> models.QuerySet[FieldMappingRule]:
+        return super().get_queryset(request).select_related("profile__parent", "created_by")

src/country_workspace/admin/program.py

@@ -64,6 +64,12 @@ def population(self, btn: "LinkButton") -> None:
         obj = btn.context["original"]
         btn.href = f"{base}?program__exact={obj.pk}&country_office__exact={obj.country_office.pk}"
 
+    @link(change_list=False)
+    def mapping_profiles(self, btn: "LinkButton") -> None:
+        obj = btn.context["original"]
+        base = reverse("admin:country_workspace_mappingprofile_changelist")
+        btn.href = f"{base}?program__id__exact={obj.pk}"
+
     @button()
     def zap(self, request: HttpRequest, pk: str) -> None:
         obj: Program = self.get_object(request, pk)

src/country_workspace/contrib/aurora/forms.py

@@ -3,7 +3,7 @@
 from django import forms
 
 from country_workspace.contrib.aurora.models import Registration
-from country_workspace.models import Program
+from country_workspace.models import MappingProfile, Program
 
 
 class ImportAuroraForm(forms.Form):
@@ -13,17 +13,22 @@ class ImportAuroraForm(forms.Form):
         help_text="What type of registrations are being imported.",
     )
     household_column_prefix = forms.CharField(
-        initial="household_", help_text="Household's column group prefix", required=False
+        initial="household-info_", help_text="Household's column group prefix", required=False
     )
     individuals_column_prefix = forms.CharField(
-        initial="individuals_",
+        initial="individual-details_",
         help_text="Individuals' column group prefix",
     )
     household_label_column = forms.CharField(
         required=False,
         initial="family_name",
         help_text="Which Individual's column should be used as label for the household.",
     )
+    mapping_profile = forms.ModelChoiceField(
+        required=False,
+        queryset=MappingProfile.objects.none(),
+        help_text="Mapping profile to use for this import. It will be used to map fields from the file to the model.",
+    )
     check_before = forms.BooleanField(
         required=False, help_text="Prevent import if errors if data is not valid against data checker."
     )
@@ -33,12 +38,24 @@ class ImportAuroraForm(forms.Form):
 
     def __init__(self, *args: Any, program: Program | None = None, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
-        self.program = program
-        if program:
-            self.fields["registration"].queryset = Registration.objects.filter(project__program=program, active=True)
-            if not (program.beneficiary_group and program.beneficiary_group.master_detail):
-                self.fields = {
-                    key: value
-                    for key, value in self.fields.items()
-                    if key not in ("household_column_prefix", "household_label_column")
-                }
+        if not program:
+            self.fields["registration"].queryset = Registration.objects.none()
+            self.fields["mapping_profile"].queryset = MappingProfile.objects.none()
+            return
+
+        is_master_detail = program.beneficiary_group and program.beneficiary_group.master_detail
+        schema = MappingProfile.ImportSchema.HH_IND if is_master_detail else MappingProfile.ImportSchema.PEOPLE
+        if not is_master_detail:
+            self.fields = {
+                key: value
+                for key, value in self.fields.items()
+                if key not in ("household_column_prefix", "household_label_column")
+            }
+
+        self.fields["registration"].queryset = Registration.objects.filter(project__program=program, active=True)
+        self.fields["mapping_profile"].queryset = MappingProfile.objects.filter(
+            source_type__in=[MappingProfile.SourceType.AURORA, MappingProfile.SourceType.ANY],
+            import_schema__in=[schema, MappingProfile.ImportSchema.ANY],
+            program=program,
+            is_active=True,
+        ).select_related("parent")

src/country_workspace/contrib/aurora/pipeline.py

@@ -1,10 +1,12 @@
+from dataclasses import dataclass, field
 from typing import Any, Final, Mapping, NotRequired
+from contextlib import suppress
 
 from django.db.transaction import atomic
 
 from country_workspace.contrib.aurora.client import AuroraClient
 from country_workspace.contrib.aurora.exceptions import TooManyBeneficiaryError
-from country_workspace.models import AsyncJob, Batch, Household, Individual
+from country_workspace.models import AsyncJob, Batch, Household, Individual, MappingProfile
 from country_workspace.utils.config import BatchNameConfig, FailIfAlienConfig
 from country_workspace.utils.fields import clean_field_names
 from country_workspace.validators.beneficiaries import validate_beneficiaries
@@ -16,66 +18,69 @@ class Config(BatchNameConfig, FailIfAlienConfig):
     household_column_prefix: NotRequired[str]
     individuals_column_prefix: str
     household_label_column: NotRequired[str]
+    mapping_profile_pk: NotRequired[int]
 
 
 RELATIONSHIP_HEAD: Final[str] = "HEAD"
 RELATIONSHIP_FIELDNAME: Final[str] = "relationship"
 
 
-def import_from_aurora(job: AsyncJob) -> dict[str, int]:
-    """Import data from the Aurora system into the database within an atomic transaction.
+@dataclass
+class AuroraImporter:
+    """Aurora data importer with mapping profile support."""
 
-    Args:
-        job (AsyncJob): The job instance containing the configuration and context for data import.
-            Expected keys in `job.config` correspond to the `Config` TypedDict.
+    job: AsyncJob
+    cfg: Config
+    client: AuroraClient = field(default_factory=AuroraClient)
+    mapping_profile: MappingProfile | None = field(init=False, default=None)
 
-    Returns:
-        dict[str, int]: Counts of imported records:
-            - "households": Number of households imported (0 if `master_detail` is False or None).
-            - "individuals": Total number of individuals imported.
+    def __post_init__(self) -> None:
+        self.batch = Batch.objects.create(
+            name=self.cfg["batch_name"],
+            program=self.job.program,
+            country_office=self.job.program.country_office,
+            imported_by=self.job.owner,
+            source=Batch.BatchSource.AURORA,
+        )
+        if self.cfg.get("mapping_profile_pk"):
+            with suppress(MappingProfile.DoesNotExist):
+                self.mapping_profile = MappingProfile.objects.get(id=self.cfg["mapping_profile_pk"], is_active=True)
 
-    Raises:
-        ValueError: If record ID is invalid or missing.
+    def run_import(self) -> dict[str, int]:
+        """Execute the Aurora import process.
 
-    """
-    with atomic():
+        Returns:
+            dict[str, int]: Counts of imported records:
+                - "households": Number of households imported.
+                - "individuals": Number of individuals imported.
+
+        """
         total = {"households": 0, "individuals": 0}
         records_data = []
-        cfg: Config = job.config
 
-        batch = Batch.objects.create(
-            name=cfg["batch_name"],
-            program=job.program,
-            country_office=job.program.country_office,
-            imported_by=job.owner,
-            source=Batch.BatchSource.AURORA,
-        )
-
-        client = AuroraClient()
-        for record in client.get(f"registration/{cfg['registration_reference_pk']}/records/"):
-            try:
-                record_id = int(record["flatten"]["id"])
-            except (ValueError, TypeError, KeyError):
-                raise ValueError(f"Invalid or missing record ID: {record.get('flatten', {}).get('id')}")
-
-            individuals = create_individuals(batch, record["flatten"], cfg)
+        for record in self.client.get(f"registration/{self.cfg['registration_reference_pk']}/records/"):
+            record_id = _extract_record_id(record)
+            individuals = create_individuals(self.batch, record["flatten"], self.cfg, self.mapping_profile)
             total["individuals"] += len(individuals)
-            if cfg["master_detail"] and individuals and individuals[0].household_id:
+            if self.cfg["master_detail"] and individuals and individuals[0].household_id:
                 total["households"] += 1
             records_data.append((record_id, individuals))
 
-        validate_records(records_data, cfg)
+        validate_records(records_data, self.cfg)
+        return total
 
-    return total
+
+def import_from_aurora(job: AsyncJob) -> dict[str, int]:
+    """Import data from the Aurora system into the database within an atomic transaction."""
+    with atomic():
+        cfg: Config = job.config
+        importer = AuroraImporter(job=job, cfg=cfg)
+        return importer.run_import()
 
 
 def validate_records(records_data: list[tuple[int, list[Individual]]], cfg: Config) -> None:
     """Validate beneficiaries based on configuration and record data.
 
-    Args:
-        records_data: List of tuples containing record ID and created individuals.
-        cfg: Configuration for validation and mapping.
-
     Raises:
         TooManyBeneficiaryError: If more than one Individual is created when master_detail is False.
 
@@ -95,14 +100,11 @@ def validate_records(records_data: list[tuple[int, list[Individual]]], cfg: Conf
         validate_beneficiaries(cfg, mapping)
 
 
-def create_household(batch: Batch, data: dict[str, Any], prefix: str) -> Household:
+def create_household(
+    batch: Batch, data: dict[str, Any], prefix: str, mapping_profile: MappingProfile | None = None
+) -> Household:
     """Create a Household object from the provided data and associate it with a batch.
 
-    Args:
-        batch (Batch): The batch to which the household will be linked.
-        data (dict[str, Any]): A dictionary containing household-related information.
-        prefix (str): The prefix used to filter and group household-related information.
-
     Returns:
         Household: The newly created household instance.
 
@@ -114,22 +116,21 @@ def create_household(batch: Batch, data: dict[str, Any], prefix: str) -> Househo
     count = len(hh_data)
     if count > 1:
         raise TooManyBeneficiaryError("Household", record_id=data["id"], count=count)
-    flex_fields = clean_field_names(next(iter(hh_data.values()), {}))
+
+    raw_fields = clean_field_names(next(iter(hh_data.values()), {}))
+    flex_fields = mapping_profile.apply_all_rules(raw_fields) if mapping_profile else raw_fields
+
     return batch.program.households.create(batch=batch, flex_fields=flex_fields)
 
 
 def create_individuals(
     batch: Batch,
     data: dict[str, Any],
     cfg: Config,
+    mapping_profile: MappingProfile | None = None,
 ) -> list[Individual]:
     """Create and associate Individual objects with an optional Household.
 
-    Args:
-        batch (Batch): The batch to which individuals will be linked.
-        data (dict[str, Any]): A dictionary containing related information.
-        cfg (Config): Configuration dictionary containing various settings for the import process.
-
     Returns:
         list[Individual]: A list of successfully created Individual instances.
 
@@ -140,10 +141,11 @@ def create_individuals(
     inds_data = _collect_by_prefix(data, cfg.get("individuals_column_prefix"))
 
     if inds_data and cfg["master_detail"] and (hh_prefix := cfg.get("household_column_prefix")):
-        household = create_household(batch, data, hh_prefix)
+        household = create_household(batch, data, hh_prefix, mapping_profile)
 
     for ind_data in inds_data.values():
-        flex_fields = clean_field_names(ind_data)
+        cleaned_data = clean_field_names(ind_data)
+        flex_fields = mapping_profile.apply_all_rules(cleaned_data) if mapping_profile else cleaned_data
         if household and (hh_label := cfg.get("household_label_column")) and not head_found:
             head_found = _update_household_label_from_individual(household, flex_fields, hh_label)
         individuals.append(
@@ -157,6 +159,19 @@ def create_individuals(
     return batch.program.individuals.bulk_create(individuals, batch_size=1000)
 
 
+def _extract_record_id(record: dict[str, Any]) -> int:
+    """Extract and validate record ID from Aurora record.
+
+    Raises:
+        ValueError: If record ID is invalid or missing.
+
+    """
+    try:
+        return int(record["flatten"]["id"])
+    except (ValueError, TypeError, KeyError):
+        raise ValueError(f"Invalid or missing record ID: {record.get('flatten', {}).get('id')}")
+
+
 def _collect_by_prefix(data: dict[str, Any], prefix: str) -> dict[str, dict[str, Any]]:
     """Extract and group fields from a dictionary based on a given prefix.
 
@@ -197,13 +212,6 @@ def _update_household_label_from_individual(
 ) -> bool:
     """Update the household's name based on an individual's role and specified name field.
 
-    Args:
-        household (Household): The household instance to update.
-        ind_data (dict[str, Any]): A dictionary containing the individual's data,
-            including relationship status and potential household name.
-        household_label_column (str): The key in the individual's data that stores
-            the name to assign to the household.
-
     Returns:
         bool: True if the household name was updated (individual is head and name provided), False otherwise.