Feature/default values for flexfields (#276)

feat: add configurable default beneficiary fields
- Add JSONField system_fields to Program model for internal metadata
- Add helpers to manage default fields (backed by system_fields["default_fields"], configurable via admin UI)
- Integrate default fields into import pipelines (Aurora, Kobo, RDI/Excel)

Author: vitali-yanushchyk-valor

src/country_workspace/contrib/aurora/import_processing.py

@@ -88,6 +88,7 @@ def create_people(batch: Batch, record: dict[str, Any], config: Config) -> Indiv
         clean_field_names,
         partial(batch.program.apply_mapping_importer, Individual),
         make_full_name,
+        partial(batch.program.apply_default_fields, Individual),
     )
     return Individual.objects.create(
         batch_id=batch.pk,

src/country_workspace/contrib/kobo/sync.py

@@ -71,9 +71,14 @@ def normalize_json(data: dict[str, Any]) -> dict[str, Any]:
 type Raw = dict[str, Any]
 
 
-def preprocess(raw: Raw, fields_to_uppercase: tuple[str, ...], mapping_importer: Callable[[Raw], Raw]) -> Raw:
+def preprocess(
+    raw: Raw,
+    fields_to_uppercase: tuple[str, ...],
+    mapping_importer: Callable[[Raw], Raw],
+    default_fields_applier: Callable[[Raw], Raw],
+) -> Raw:
     clean: Callable[[Raw], Raw] = partial(clean_field_names, fields_to_uppercase=fields_to_uppercase)
-    processor: Callable[[Raw], Raw] = compose(normalize_json, clean, mapping_importer)
+    processor: Callable[[Raw], Raw] = compose(normalize_json, clean, mapping_importer, default_fields_applier)
     return processor(raw)
 
 
@@ -88,6 +93,7 @@ def create_individuals(batch: Batch, household: Household, submission: Submissio
             raw_individual,
             INDIVIDUAL_FIELDS_TO_UPPERCASE + TO_UPPERCASE_FIELDS,
             partial(batch.program.apply_mapping_importer, Individual),
+            partial(batch.program.apply_default_fields, Individual),
         )
         fullname = get_fullname_key(individual_fields)
         individuals.append(
@@ -111,6 +117,7 @@ def create_household(
         raw_household_fields,
         HOUSEHOLD_FIELDS_TO_UPPERCASE,
         partial(batch.program.apply_mapping_importer, Household),
+        partial(batch.program.apply_default_fields, Household),
     )
     household_fields["household_id"] = id_generator()
     return cast(

src/country_workspace/datasources/rdi/processors.py

@@ -2,7 +2,7 @@
 from base64 import b64encode
 from collections import defaultdict
 from collections.abc import Generator
-from copy import deepcopy
+from functools import partial
 from typing import Any, cast, Mapping
 
 from PIL import Image
@@ -18,6 +18,7 @@
 from country_workspace.utils.imports import validate_alien_fields
 from country_workspace.utils.functional import compose
 from country_workspace.workspaces.admin.cleaners.validate import create_validation_jobs
+
 from .config import Config, SheetName, Sheet
 from .exceptions import ColumnConfigurationError, SheetProcessingError, SheetNotFoundError
 from .utils import date_to_iso_string, datetime_to_date
@@ -108,24 +109,23 @@ 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 = {}
+    transform_row = compose(
+        clean_field_names,
+        partial(job.program.apply_mapping_importer, Household),
+        partial(job.program.apply_default_fields, Household),
+    )
 
     for row in sheet:
-        household_key = get_value(row, config["household_id_column"])
-        if household_key in mapping:
+        if (household_key := get_value(row, config["household_id_column"])) in mapping:
             raise SheetProcessingError(SheetName.HOUSEHOLDS, household_key)
-
-        label = get_value(row, config["household_label"])
-        raw_data = clean_field_names(row)
-        flex_fields = job.program.apply_mapping_importer(Household, deepcopy(raw_data))
-
         try:
             mapping[household_key] = cast(
                 "Household",
                 Household.objects.create(
                     batch_id=batch.pk,
-                    name=str(label),
-                    flex_fields=flex_fields,
-                    raw_data=raw_data,
+                    name=str(get_value(row, config["household_label"])),
+                    flex_fields=transform_row(row),
+                    raw_data=row,
                 ),
             )
         except Exception as e:
@@ -141,6 +141,11 @@ 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
+    transform_row = compose(
+        clean_field_names,
+        partial(job.program.apply_mapping_importer, Individual),
+        partial(job.program.apply_default_fields, Individual),
+    )
 
     for row in sheet:
         beneficiary_key = get_value(row, config["beneficiary_id_column"])
@@ -150,18 +155,15 @@ 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)
-        raw_data = clean_field_names(cleaned_row)
-        flex_fields = job.program.apply_mapping_importer(Individual, deepcopy(raw_data))
-
         try:
             mapping[beneficiary_key] = cast(
                 "Individual",
                 Individual.objects.create(
                     batch_id=batch.pk,
                     name=name,
                     household=household,
-                    flex_fields=flex_fields,
-                    raw_data=raw_data,
+                    flex_fields=transform_row(cleaned_row),
+                    raw_data=row,
                 ),
             )
         except Exception as e:

src/country_workspace/migrations/0032_program_system_fields.py

@@ -0,0 +1,21 @@
+# Generated by Django 5.2.3 on 2025-11-20 10:28
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("country_workspace", "0031_add_parse_name_permission"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="program",
+            name="system_fields",
+            field=models.JSONField(
+                blank=True,
+                default=dict,
+                help_text='Internal metadata (e.g. "default_fields" for household/individual defaults).',
+            ),
+        ),
+    ]

src/country_workspace/models/program.py

@@ -1,6 +1,7 @@
 from contextlib import suppress
-from typing import TYPE_CHECKING
+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
@@ -23,6 +24,11 @@
     from .individual import Individual
 
 
+class BeneficiaryScope(StrEnum):
+    HOUSEHOLD = "household"
+    INDIVIDUAL = "individual"
+
+
 class Program(BaseModel):
     DRAFT = "DRAFT"
     ACTIVE = "ACTIVE"
@@ -95,6 +101,17 @@ class Program(BaseModel):
     extra_fields = models.JSONField(default=dict, blank=True, null=False)
     enabled = models.BooleanField(default=True, db_index=True, help_text="Is this program enabled in the workspace?")
     serializer = models.ForeignKey(DataSerializer, on_delete=models.SET_NULL, null=True, blank=True)
+    # Internal metadata used by the workspace.
+    # Expected JSON structure (simplified):
+    # - key "default_fields": an object with two optional keys:
+    #   - "household":  mapping "<field_name>" -> <default_value>
+    #   - "individual": mapping "<field_name>" -> <default_value>
+    # - other keys may be added in the future.
+    system_fields = models.JSONField(
+        default=dict,
+        blank=True,
+        help_text=_('Internal metadata (e.g. "default_fields" for household/individual defaults).'),
+    )
 
     def __str__(self) -> str:
         return self.name
@@ -116,26 +133,26 @@ def individuals(self) -> "QuerySet[Individual]":
 
         return Individual.objects.filter(batch__program=self)
 
-    def get_checker_for(self, m: type[Validable] | Validable) -> DataChecker:
-        from country_workspace.models import Household, Individual
-        from country_workspace.workspaces.models import CountryHousehold, CountryIndividual
-
-        if isinstance(m, (Household | CountryHousehold)) or m in (Household, CountryHousehold):
-            return self.household_checker
-        if isinstance(m, (Individual | CountryIndividual)) or m in (Individual, CountryIndividual):
-            return self.individual_checker
-        raise ValueError(m)
-
-    def get_columns_for(self, model_cls: type[Validable]) -> list[str]:
+    @staticmethod
+    def _scope_for(m: type[Validable] | Validable) -> BeneficiaryScope:
+        """Return 'household' or 'individual' for a given model class or instance."""
         from country_workspace.models import Household, Individual
 
-        if issubclass(model_cls, Household):
-            raw = self.household_columns
-        elif issubclass(model_cls, Individual):
-            raw = self.individual_columns
-        else:
-            raise TypeError(f"Unsupported model {model_cls!r}")
-
+        cls = m if isinstance(m, type) else type(m)
+        if issubclass(cls, Household):
+            return BeneficiaryScope.HOUSEHOLD
+        if issubclass(cls, Individual):
+            return BeneficiaryScope.INDIVIDUAL
+        raise TypeError(f"Unsupported model {m!r}")
+
+    def get_checker_for(self, m: type[Validable] | Validable) -> DataChecker | None:
+        scope = self._scope_for(m)
+        return getattr(self, f"{scope.value}_checker")
+
+    def get_columns_for(self, m: type[Validable]) -> list[str]:
+        """Return list of column names for the given model class."""
+        scope = self._scope_for(m)
+        raw = getattr(self, f"{scope.value}_columns")
         return [c.strip() for c in raw.splitlines() if c.strip()]
 
     def serialize(self, data: list[dict]) -> Iterable:
@@ -146,7 +163,38 @@ def serialize(self, data: list[dict]) -> Iterable:
     def apply_mapping_importer(
         self, m: type[Validable] | Validable, data: dict[str, str | int | bool]
     ) -> dict[str, str | int | bool]:
-        # skip if mapping importer not found
+        """Apply mapping importer from the checker's mappingimporter, if any."""
+        if (checker := self.get_checker_for(m)) is None:
+            return data
         with suppress(ObjectDoesNotExist):
-            self.get_checker_for(m).mappingimporter.apply(data)
+            checker.mappingimporter.apply(data)
+        return data
+
+    def get_default_fields_for(self, m: type[Validable] | Validable) -> dict[str, Any]:
+        """Return defaults from system_fields['default_fields'][scope] for the given model."""
+        scope = self._scope_for(m).value
+        default_fields = (self.system_fields or {}).get("default_fields") or {}
+        raw = default_fields.get(scope) or {}
+        return dict(raw) if isinstance(raw, dict) else {}
+
+    def save_default_fields_for(self, m: type[Validable] | Validable, defaults: dict[str, Any]) -> None:
+        """Replace system_fields['default_fields'][scope] with the given defaults and save."""
+        scope = self._scope_for(m).value
+
+        system_fields = dict(self.system_fields or {})
+        default_fields = dict(system_fields.get("default_fields") or {})
+
+        default_fields[scope] = defaults
+        system_fields["default_fields"] = default_fields
+
+        self.system_fields = system_fields
+        self.save(update_fields=["system_fields"])
+
+    def apply_default_fields(self, m: type[Validable] | Validable, data: dict[str, Any]) -> dict[str, Any]:
+        """Apply default fields for the given model to the provided data dict."""
+        if not (defaults := self.get_default_fields_for(m)):
+            return data
+        for field_name, default_value in defaults.items():
+            if field_name not in data or data[field_name] is None:
+                data[field_name] = default_value
         return data

src/country_workspace/workspaces/admin/forms.py

@@ -124,3 +124,12 @@ def __init__(self, *args: Any, beneficiary_group: BeneficiaryGroup | None = None
 
         [self.fields.pop(field_name, None) for field_name in fields_to_exclude]
         [setattr(self.fields["beneficiary_id_column"], attr, v) for attr, v in field_cfg.items()]
+
+
+class MassDefaultsForm(forms.Form):
+    def __init__(self, *args: Any, checker: "DataChecker", **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        source_form = checker.get_form()()
+        for field in source_form.fields.values():
+            field.required = False
+        self.fields.update(source_form.fields)