fix(bulk-upload): update error writer logic to write error details on csv

Author: sudip-khanal

local_units/bulk_upload.py

@@ -6,6 +6,7 @@
 
 from django.core.files.base import ContentFile
 from django.db import transaction
+from rest_framework.exceptions import ErrorDetail
 from rest_framework.serializers import Serializer
 
 from local_units.models import HealthData, LocalUnit, LocalUnitBulkUpload, LocalUnitType
@@ -27,16 +28,69 @@ class BulkUploadError(Exception):
 class ErrorWriter:
     def __init__(self, fieldnames: list[str]):
         self._fieldnames = ["upload_status"] + fieldnames
+        self._rows: list[dict[str, str]] = []
         self._output = io.StringIO()
         self._writer = csv.DictWriter(self._output, fieldnames=self._fieldnames)
         self._writer.writeheader()
+        self._has_errors = False
+
+    def _format_errors(self, errors: dict) -> dict[str, list[str]]:
+        """Format serializer errors into field_name and list of messages."""
+        error = {}
+        for key, value in errors.items():
+            if isinstance(value, dict):
+                for inner_key, inner_value in self._format_errors(value).items():
+                    error[inner_key] = inner_value
+            elif isinstance(value, list):
+                error[key] = [self._clean_message(v) for v in value]
+            else:
+                error[key] = [self._clean_message(value)]
+        return error
+
+    def _clean_message(self, msg: Any) -> str:
+        """Convert ErrorDetail or other objects into normal text."""
+        if isinstance(msg, ErrorDetail):
+            return str(msg)
+        return str(msg)
+
+    def _update_csv_header_with_errors(self):
+        """Update the CSV with updated headers when new error columns are introduced."""
+        self._output.seek(0)
+        self._output.truncate()
+        self._writer = csv.DictWriter(self._output, fieldnames=self._fieldnames)
+        self._writer.writeheader()
+        for row in self._rows:
+            self._writer.writerow(row)
 
     def write(
-        self, row: dict[str, str], status: Literal[LocalUnitBulkUpload.Status.SUCCESS, LocalUnitBulkUpload.Status.FAILED]
+        self,
+        row: dict[str, str],
+        status: Literal[LocalUnitBulkUpload.Status.SUCCESS, LocalUnitBulkUpload.Status.FAILED],
+        error_detail: dict | None = None,
     ) -> None:
-        self._writer.writerow({"upload_status": status.name, **row})
-        if status == LocalUnitBulkUpload.Status.FAILED:
-            self._has_errors = True
+        row_copy = {col: row.get(col, "") for col in self._fieldnames}
+        row_copy["upload_status"] = status.name
+        added_error_column = False
+
+        if status == LocalUnitBulkUpload.Status.FAILED and error_detail:
+            formatted_errors = self._format_errors(error_detail)
+            for field, messages in formatted_errors.items():
+                error_col = f"{field}_error"
+
+                if error_col not in self._fieldnames:
+                    if field in self._fieldnames:
+                        col_idx = self._fieldnames.index(field)
+                        self._fieldnames.insert(col_idx + 1, error_col)
+                    else:
+                        self._fieldnames.append(error_col)
+
+                    added_error_column = True
+                row_copy[error_col] = "; ".join(messages)
+        self._rows.append(row_copy)
+        if added_error_column:
+            self._update_csv_header_with_errors()
+        else:
+            self._writer.writerow(row_copy)
 
     def to_content_file(self) -> ContentFile:
         return ContentFile(self._output.getvalue().encode("utf-8"))
@@ -70,17 +124,17 @@ def process_row(self, data: Dict[str, Any]) -> bool:
         if serializer.is_valid():
             self.bulk_manager.add(LocalUnit(**serializer.validated_data))
             return True
+        self.error_detail = serializer.errors
         return False
 
     def run(self) -> None:
         with self.bulk_upload.file.open("rb") as csv_file:
             file = io.TextIOWrapper(csv_file, encoding="utf-8")
             csv_reader = csv.DictReader(file)
             fieldnames = csv_reader.fieldnames or []
-
             try:
                 self._validate_csv(fieldnames)
-            except ValueError as e:
+            except BulkUploadError as e:
                 self.bulk_upload.status = LocalUnitBulkUpload.Status.FAILED
                 self.bulk_upload.error_message = str(e)
                 self.bulk_upload.save(update_fields=["status", "error_message"])
@@ -99,7 +153,9 @@ def run(self) -> None:
                             self.error_writer.write(row_data, status=LocalUnitBulkUpload.Status.SUCCESS)
                         else:
                             self.failed_count += 1
-                            self.error_writer.write(row_data, status=LocalUnitBulkUpload.Status.FAILED)
+                            self.error_writer.write(
+                                row_data, status=LocalUnitBulkUpload.Status.FAILED, error_detail=self.error_detail
+                            )
                             logger.warning(f"[BulkUpload:{self.bulk_upload.pk}] Row '{row_index}' failed")
 
                     if self.failed_count > 0:
@@ -175,7 +231,7 @@ def _validate_csv(self, fieldnames) -> None:
             raise ValueError("Invalid local unit type")
 
         if present_health_fields and local_unit_type.name.strip().lower() != "health care":
-            raise ValueError(f"Health data are not allowed for this type: {local_unit_type.name}.")
+            raise BulkUploadError(f"Health data are not allowed for this type: {local_unit_type.name}.")
 
 
 class BulkUploadHealthData(BaseBulkUpload[LocalUnitUploadContext]):

local_units/serializers.py

@@ -905,7 +905,7 @@ def validate_level(self, value):
         level_name = value.strip().lower()
         level_id = self.level_map.get(level_name)
         if not level_id:
-            raise serializers.ValidationError({"Level": gettext("Level '{value}' is not valid")})
+            raise serializers.ValidationError(gettext("Level is not valid"))
         return level_id
 
     def validate(self, validated_data):

local_units/tasks.py

@@ -138,7 +138,7 @@ def process_bulk_upload_local_unit(bulk_upload_id: int) -> None:
     bulk_upload: LocalUnitBulkUpload | None = LocalUnitBulkUpload.objects.filter(id=bulk_upload_id).first()
 
     if not bulk_upload:
-        logger.error(f"BulkUploadLocalUnit:'{bulk_upload_id}' Not found.", exc_info=True)
+        logger.warning(f"BulkUploadLocalUnit:'{bulk_upload_id}' Not found.", exc_info=True)
         return
     try:
         if (
@@ -149,6 +149,6 @@ def process_bulk_upload_local_unit(bulk_upload_id: int) -> None:
             BaseBulkUploadLocalUnit(bulk_upload).run()
 
     except Exception as exc:
-        logger.error(f"BulkUploadLocalUnit:'{bulk_upload_id}' Failed with exception: {exc}", exc_info=True)
+        logger.warning(f"BulkUploadLocalUnit:'{bulk_upload_id}' Failed with exception: {exc}", exc_info=True)
         bulk_upload.update_status(LocalUnitBulkUpload.Status.FAILED)
         raise exc