Use ValidatorMixin.validate fail_if_alien argument

Author: sergey-misuk-personal

src/country_workspace/contrib/aurora/pipeline.py

@@ -5,7 +5,7 @@
 from country_workspace.contrib.aurora.client import AuroraClient
 from country_workspace.models import AsyncJob, Batch, Household, Individual
 from country_workspace.utils.config import BatchNameConfig, FailIfAlienConfig
-from country_workspace.utils.fields import uppercase_field_value, RecordPreprocessor, create_json_record_preprocessor
+from country_workspace.utils.fields import uppercase_field_value, clean_field_names
 
 
 class Config(BatchNameConfig, FailIfAlienConfig):
@@ -43,38 +43,30 @@ def import_from_aurora(job: AsyncJob) -> dict[str, int]:
         source=Batch.BatchSource.AURORA,
     )
     client = AuroraClient()
-    individual_preprocessor = create_json_record_preprocessor(config, job.program.individual_checker)
-    household_preprocessor = create_json_record_preprocessor(config, job.program.household_checker)
     with atomic():
         for record in client.get(f"registration/{config['registration_reference_pk']}/records/"):
             inds_data = _collect_by_prefix(record["flatten"], config.get("individuals_column_prefix"))
             if inds_data:
-                hh = create_household(
-                    batch, record["flatten"], config.get("household_column_prefix"), household_preprocessor
-                )
+                hh = create_household(batch, record["flatten"], config.get("household_column_prefix"))
                 total_hh += 1
                 total_ind += len(
                     create_individuals(
                         household=hh,
                         data=inds_data,
                         household_label_column=config.get("household_label_column"),
-                        preprocess_record=individual_preprocessor,
                     )
                 )
     return {"households": total_hh, "individuals": total_ind}
 
 
-def create_household(
-    batch: Batch, data: dict[str, Any], prefix: str, preprocess_record: RecordPreprocessor
-) -> Household:
+def create_household(batch: Batch, data: dict[str, Any], prefix: str) -> 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.
-        preprocess_record (RecordPreprocessor): The function normalizing field names and checking if they are valid.
 
     Returns:
         Household: The newly created household instance.
@@ -87,19 +79,16 @@ def create_household(
     if len(flex_fields) > 1:
         raise ValueError("Multiple households found")
     flex_fields = next(iter(flex_fields.values()), {})
-    return batch.program.households.create(batch=batch, flex_fields=preprocess_record(flex_fields))
+    return batch.program.households.create(batch=batch, flex_fields=clean_field_names(flex_fields))
 
 
-def create_individuals(
-    household: Household, data: dict[str, Any], household_label_column: str, preprocess_record: RecordPreprocessor
-) -> list[Individual]:
+def create_individuals(household: Household, data: dict[str, Any], household_label_column: str) -> list[Individual]:
     """Create and associate Individual objects with a given Household.
 
     Args:
         household (Household): The household to which the individuals will be linked.
         data (dict[str, Any]): A dictionary mapping indices to individual details.
         household_label_column (str): The key in the individual data used to determine the household label.
-        preprocess_record (RecordPreprocessor): The function normalizing field names and checking if they are valid.
 
     Returns:
         list[Individual]: A list of successfully created Individual instances.
@@ -109,7 +98,7 @@ def create_individuals(
     head_found = False
 
     for raw_individual in data.values():
-        individual = preprocess_record(raw_individual)
+        individual = clean_field_names(raw_individual)
         if not head_found:
             head_found = _update_household_label_from_individual(household, individual, household_label_column)
         individuals.append(

src/country_workspace/contrib/kobo/sync.py

@@ -7,9 +7,9 @@
 from country_workspace.contrib.kobo.api.data.asset import Asset
 from country_workspace.contrib.kobo.api.data.submission import Submission
 from country_workspace.contrib.kobo.models import KoboSubmission
-from country_workspace.models import AsyncJob, Batch, Household, Individual, Program
+from country_workspace.models import AsyncJob, Batch, Household, Individual
 from country_workspace.utils.config import FailIfAlienConfig, BatchNameConfig
-from country_workspace.utils.fields import create_json_record_preprocessor, RecordPreprocessor
+from country_workspace.utils.fields import clean_field_names
 
 
 class Config(BatchNameConfig, FailIfAlienConfig):
@@ -32,9 +32,7 @@ def extract_household_data(submission: Submission, individual_records_field: str
     return {key: value for key, value in submission.items() if key != individual_records_field}
 
 
-def create_individuals(
-    batch: Batch, household: Household, submission: Submission, config: Config, preprocess_record: RecordPreprocessor
-) -> int:
+def create_individuals(batch: Batch, household: Household, submission: Submission, config: Config) -> int:
     individuals = []
     for raw_individual in submission.get(config["individual_records_field"], []):
         individual = {
@@ -46,22 +44,20 @@ def create_individuals(
                 batch=batch,
                 household=household,
                 name=individual.get(fullname, ""),
-                flex_fields=preprocess_record(individual),
+                flex_fields=clean_field_names(individual),
             ),
         )
     household.program.individuals.bulk_create(individuals)
     return len(individuals)
 
 
-def create_household(
-    batch: Batch, submission: Submission, config: Config, preprocess_record: RecordPreprocessor
-) -> Household:
+def create_household(batch: Batch, submission: Submission, config: Config) -> Household:
     household_fields = extract_household_data(submission, config["individual_records_field"])
     return cast(
         Household,
         batch.program.households.create(
             batch=batch,
-            flex_fields=preprocess_record(household_fields),
+            flex_fields=clean_field_names(household_fields),
         ),
     )
 
@@ -74,21 +70,18 @@ class ImportResult(TypedDict):
     individuals: int
 
 
-def import_asset(batch: Batch, asset: Asset, config: Config, program: Program) -> ImportResult:
+def import_asset(batch: Batch, asset: Asset, config: Config) -> ImportResult:
     household_counter = 0
     individual_counter = 0
 
-    individual_preprocessor = create_json_record_preprocessor(config, program.individual_checker)
-    household_preprocessor = create_json_record_preprocessor(config, program.household_checker)
-
     with cache.lock(ASSET_CACHE_KEY.format(asset_id=asset.uid)):
         submission_ids = set(KoboSubmission.objects.filter(asset_uid=asset.uid).values_list("submission_id", flat=True))
         for submission in asset.submissions:
             if submission.id in submission_ids:
                 continue
-            household = create_household(batch, submission, config, household_preprocessor)
+            household = create_household(batch, submission, config)
             household_counter += 1
-            individual_counter += create_individuals(batch, household, submission, config, individual_preprocessor)
+            individual_counter += create_individuals(batch, household, submission, config)
 
     return ImportResult(households=household_counter, individuals=individual_counter)
 
@@ -111,7 +104,7 @@ def import_data(job: AsyncJob) -> ImportResult:
     for asset in client.assets:
         # TODO: fetch specific asset
         if config["project_id"] == asset.uid:
-            import_result = import_asset(batch, asset, config, job.program)
+            import_result = import_asset(batch, asset, config)
             household_counter += import_result["households"]
             individual_counter += import_result["individuals"]
 

src/country_workspace/datasources/rdi.py

@@ -7,7 +7,7 @@
 
 from country_workspace.models import AsyncJob, Batch, Household
 from country_workspace.utils.config import FailIfAlienConfig, BatchNameConfig
-from country_workspace.utils.fields import create_json_record_preprocessor, Record
+from country_workspace.utils.fields import clean_field_names, Record
 
 RDI = str | io.BytesIO
 Sheet = Iterable[Record]
@@ -80,21 +80,17 @@ def has_household_pk(row: Record) -> bool:
 def process_households(sheet: Sheet, job: AsyncJob, batch: Batch, config: Config) -> Mapping[int, Household]:
     mapping = {}
 
-    preprocess_json_record = create_json_record_preprocessor(config, job.program.household_checker)
-
     for i, row in enumerate(sheet, 1):
         name = get_value(row, config["master_column_label"])
         household_key = get_value(row, config["household_pk_col"])
 
-        preprocessed_row = preprocess_json_record(row)
-
         try:
             mapping[household_key] = cast(
                 Household,
                 job.program.households.create(
                     batch=batch,
                     name=name,
-                    flex_fields=preprocessed_row,
+                    flex_fields=clean_field_names(row),
                 ),
             )
         except Exception as e:
@@ -108,8 +104,6 @@ def process_individuals(
 ) -> int:
     processed = 0
 
-    preprocess_json_record = create_json_record_preprocessor(config, job.program.individual_checker)
-
     for i, row in enumerate(sheet, 1):
         name = get_value(row, config["detail_column_label"])
         household_key = get_value(row, config["household_pk_col"])
@@ -118,14 +112,12 @@ def process_individuals(
         if not household:
             raise MissingHouseholdError(i, household_key)
 
-        preprocessed_row = preprocess_json_record(row)
-
         try:
             job.program.individuals.create(
                 batch=batch,
                 name=name,
                 household_id=household.pk,
-                flex_fields=preprocessed_row,
+                flex_fields=clean_field_names(row),
             )
         except Exception as e:
             raise SheetProcessingError(INDIVIDUAL, i) from e
@@ -138,7 +130,7 @@ def process_individuals(
 def validate_households(config: Config, household_mapping: Mapping[int, Household]) -> None:
     if config["check_before"]:
         for household_key, household in household_mapping.items():
-            if not household.validate_with_checker():
+            if not household.validate_with_checker(fail_if_alien=config["fail_if_alien"]):
                 raise HouseholdValidationError(household_key)
 
 

src/country_workspace/models/base.py

@@ -112,8 +112,8 @@ def save(
     def checker(self) -> "DataChecker":
         raise NotImplementedError
 
-    def validate_with_checker(self) -> bool:
-        errors = self.checker.validate([self.flex_fields])
+    def validate_with_checker(self, fail_if_alien: bool = False) -> bool:
+        errors = self.checker.validate([self.flex_fields], fail_if_alien=fail_if_alien)
         if errors:
             self.errors = errors[1]
         else:

src/country_workspace/models/household.py

@@ -36,13 +36,13 @@ def program(self) -> "Program":
     def country_office(self) -> "Office":
         return self.batch.program.country_office
 
-    def validate_with_checker(self) -> bool:
+    def validate_with_checker(self, fail_if_alien: bool = False) -> bool:
         hh_valid = True
         for ind in self.members.all():
-            if not ind.validate_with_checker():
+            if not ind.validate_with_checker(fail_if_alien=fail_if_alien):
                 hh_valid = False
         if hh_valid:
-            super().validate_with_checker()
+            super().validate_with_checker(fail_if_alien=fail_if_alien)
             errors = self.program.beneficiary_validator.validate(self)
             if errors:
                 self.errors["dct"] = errors

src/country_workspace/utils/fields.py

@@ -1,17 +1,13 @@
 from collections.abc import Callable, Mapping
-from typing import Any
 from functools import reduce
+from typing import Any
 
 from django.utils import timezone
-from hope_flex_fields.models import DataChecker
-
-from country_workspace.utils.config import FailIfAlienConfig
 
 batch_name_default: Callable[[], str] = lambda: f"Batch {timezone.now()}"
 rdi_name_default: Callable[[], str] = lambda: f"RDI to HOPE {timezone.now()}"
 
 Record = Mapping[str, Any]
-RecordPreprocessor = Callable[[Record], Record]
 
 
 TO_REMOVE = "_h_c", "_h_f", "_i_c", "_i_f"
@@ -30,30 +26,8 @@ def clean_field_name(v: str) -> str:
     return reduce(lambda name, substr: name.replace(substr, ""), TO_REMOVE, v.lower())
 
 
-class ExtraFieldInRecordError(Exception):
-    def __init__(self, *fields: str) -> None:
-        super().__init__(*fields)
-        self.fields = fields
-
-    def __str__(self) -> str:
-        return f"Extra fields found: {', '.join(self.fields)}"
-
-
-def create_json_record_preprocessor(config: FailIfAlienConfig, checker: DataChecker) -> Callable[[Record], Record]:
-    if config["fail_if_alien"]:
-        field_names = {field.name for _, field in checker.get_fields()}
-    else:
-        field_names = set()
-
-    def preprocessor(record: Record) -> Record:
-        cleaned_record = {clean_field_name(k): v for k, v in record.items()}
-
-        if config["fail_if_alien"] and (extra_fields := cleaned_record.keys() - field_names):
-            raise ExtraFieldInRecordError(*extra_fields)
-
-        return cleaned_record
-
-    return preprocessor
+def clean_field_names(record: Record) -> Record:
+    return {clean_field_name(k): v for k, v in record.items()}
 
 
 def uppercase_field_value(k: str, v: Any) -> str: