feat(bulk-upload): add .xlsm file support to bulk upload

Author: sudip-khanal

go-static/files/local_units/Health Care Bulk Import Template - Local Units.xlsx

@@ -115,6 +115,7 @@ def to_content_file(self) -> ContentFile:
 
 class BaseBulkUpload(Generic[ContextType]):
     serializer_class: Type[Serializer]
+    HEADER_MAP: dict[str, str]
 
     def __init__(self, bulk_upload: LocalUnitBulkUpload):
         if self.serializer_class is None:
@@ -152,70 +153,71 @@ def process_row(self, data: Dict[str, Any]) -> bool:
         return False
 
     def run(self) -> None:
-        with self.bulk_upload.file.open("rb") as f:
-            try:
-                # TODO(sudip): Use read_only while reading xlsx file
-                workbook = openpyxl.load_workbook(f, data_only=True)
+        header_row_index = 2
+        data_row_index = header_row_index + 2
+
+        try:
+            with self.bulk_upload.file.open("rb") as f:
+                workbook = openpyxl.load_workbook(f, data_only=True, read_only=True)
                 sheet = workbook.active
-                header_row_index = 2
-                data_row_index = header_row_index + 2
 
-                # Read header row
                 headers = next(sheet.iter_rows(values_only=True, min_row=header_row_index, max_row=header_row_index))
                 raw_fieldnames = [str(h).strip() for h in headers if h and str(h).strip()]
-                header_map = getattr(self, "HEADER_MAP", {}) or {}
+                header_map = self.HEADER_MAP or {}
                 mapped_fieldnames = [header_map.get(h, h) for h in raw_fieldnames]
                 fieldnames = mapped_fieldnames
 
                 if self.is_excel_data_empty(sheet, data_start_row=data_row_index):
-                    raise BulkUploadError("The uploaded Excel file is empty. Please provide at least one data row.")
+                    raise BulkUploadError("The uploaded file is empty. Please provide at least one data row.")
 
                 self._validate_type(fieldnames)
-                data_rows = (
-                    row
-                    for row in sheet.iter_rows(values_only=True, min_row=4)
-                    if any(cell is not None for cell in row)  # skip the empty rows
-                )
-            except Exception 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"])
-                logger.warning(f"[BulkUpload:{self.bulk_upload.pk}] Validation error: {str(e)}")
-                return
-
-        context = self.get_context().__dict__
-        self.error_writer = ErrorWriter(fieldnames=raw_fieldnames, header_map=header_map)
-
-        try:
-            with transaction.atomic():
-                self.delete_existing_local_unit()
-
-                for row_index, row_values in enumerate(data_rows, start=data_row_index):
-                    row_dict = dict(zip(fieldnames, row_values))
-                    row_dict = {**row_dict, **context}
-                    # Convert datetime objects to strings
-                    for key, value in row_dict.items():
-                        if isinstance(value, (datetime, date)):
-                            row_dict[key] = value.strftime("%Y-%m-%d")
-                    if self.process_row(row_dict):
-                        self.success_count += 1
-                        self.error_writer.write(row_dict, status=LocalUnitBulkUpload.Status.SUCCESS)
-                    else:
-                        self.failed_count += 1
-                        self.error_writer.write(
-                            row_dict,
-                            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:
-                    raise BulkUploadError("Bulk upload failed with some errors.")
-
-                self.bulk_manager.done()
-                self._finalize_success()
-
-        except BulkUploadError:
+                self.error_writer = ErrorWriter(fieldnames=raw_fieldnames, header_map=header_map)
+                context = self.get_context().__dict__
+
+                with transaction.atomic():
+                    self.delete_existing_local_unit()
+
+                    for row_index, row_values in enumerate(
+                        sheet.iter_rows(values_only=True, min_row=data_row_index),
+                        start=data_row_index,
+                    ):
+                        if not any(cell is not None for cell in row_values):
+                            continue  # skip empty rows
+
+                        row_dict = dict(zip(fieldnames, row_values))
+                        row_dict = {**row_dict, **context}
+
+                        # Convert date/datetime to str
+                        for key, value in row_dict.items():
+                            if isinstance(value, (datetime, date)):
+                                row_dict[key] = value.strftime("%Y-%m-%d")
+
+                        if self.process_row(row_dict):
+                            self.success_count += 1
+                            self.error_writer.write(row_dict, status=LocalUnitBulkUpload.Status.SUCCESS)
+                        else:
+                            self.failed_count += 1
+                            self.error_writer.write(
+                                row_dict,
+                                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:
+                        raise BulkUploadError()
+
+                    self.bulk_manager.done()
+                    self._finalize_success()
+
+                workbook.close()
+
+        except Exception 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"])
+            if isinstance(e, BulkUploadError):
+                logger.warning(f"[BulkUpload:{self.bulk_upload.pk}]  error: {e}")
             self._finalize_failure()
 
     def _finalize_success(self) -> None:

local_units/bulk_upload.py

@@ -673,6 +673,7 @@ def update(self, instance, validated_data):
 
 
 class LocalUnitBulkUploadSerializer(serializers.ModelSerializer):
+    VALID_FILE_EXTENSIONS = (".xlsx", ".xlsm")
     country = serializers.PrimaryKeyRelatedField(
         queryset=Country.objects.filter(
             is_deprecated=False, independent=True, iso3__isnull=False, record_type=CountryType.COUNTRY
@@ -703,8 +704,8 @@ class Meta:
         )
 
     def validate_file(self, file):
-        if not file.name.endswith(".xlsx"):
-            raise serializers.ValidationError(gettext("File must be a xlsx file."))
+        if not file.name.lower().endswith(self.VALID_FILE_EXTENSIONS):
+            raise serializers.ValidationError(gettext("The uploaded file must be an Excel document (.xlsx or .xlsm)."))
         if file.size > 10 * 1024 * 1024:
             raise serializers.ValidationError(gettext("File must be less than 10 MB."))
         return file

local_units/serializers.py

@@ -1535,12 +1535,12 @@ def setUpTestData(cls):
         cls.professional_training_facilities = ProfessionalTrainingFacility.objects.create(code=1, name="Nurses")
         cls.general_medical_services = GeneralMedicalService.objects.create(code=1, name="Minor Trauma")
 
-        file_path = os.path.join(settings.TEST_DIR, "local_unit/test-health.xlsx")
+        file_path = os.path.join(settings.TEST_DIR, "local_unit/test-health.xlsm")
         with open(file_path, "rb") as f:
             cls._file_content = f.read()
 
-    def create_upload_file(cls, filename="test-health.xlsx"):
-        return SimpleUploadedFile(filename, cls._file_content, content_type="text/xlsx")
+    def create_upload_file(cls, filename="test-health.xlsm"):
+        return SimpleUploadedFile(filename, cls._file_content, content_type="text/xlsm")
 
     def test_bulk_upload_health_with_incorrect_country(cls):
         """
@@ -1565,7 +1565,7 @@ def test_bulk_upload_health_with_incorrect_country(cls):
 
     def test_bulk_upload_health_fails_and_does_not_delete(cls):
         """
-        Should fail and keep existing LocalUnits & HealthData when CSV invalid.
+        Should fail and keep existing LocalUnits & HealthData when file invalid.
         """
         health_data = HealthDataFactory.create_batch(
             5,
@@ -1644,12 +1644,12 @@ def test_empty_health_template_file(cls):
         """
 
         file_path = os.path.join(
-            settings.STATICFILES_DIRS[0], "files", "local_units", "Health Care Bulk Import Template - Local Units.xlsx"
+            settings.STATICFILES_DIRS[0], "files", "local_units", "Health Care Bulk Import Template - Local Units.xlsm"
         )
         with open(file_path, "rb") as f:
             file_content = f.read()
         empty_file = SimpleUploadedFile(
-            name="Health Care Bulk Import Template - Local Units.xlsx", content=file_content, content_type="text/xlsx"
+            name="Health Care Bulk Import Template - Local Units.xlsm", content=file_content, content_type="text/xlsm"
         )
         health_data = HealthDataFactory.create_batch(
             5,

local_units/test_views.py

@@ -436,7 +436,7 @@ class LocalUnitBulkUploadViewSet(
     def get_bulk_upload_template(self, request):
         template_type = request.query_params.get("bulk_upload_template", "local_unit")
         if template_type == "health_care":
-            file_url = request.build_absolute_uri(static("files/local_units/Health Care Bulk Import Template - Local Units.xlsx"))
+            file_url = request.build_absolute_uri(static("files/local_units/Health Care Bulk Import Template - Local Units.xlsm"))
         else:
             file_url = request.build_absolute_uri(
                 static("files/local_units/Administrative Bulk Import Template - Local Units.xlsx")