refactor(local-unit): refactor local unit status

Author: sudip-khanal

local_units/admin.py

@@ -55,13 +55,9 @@ class LocalUnitAdmin(CompareVersionAdmin, admin.OSMGeoAdmin):
         "health",
         "bulk_upload",
     )
-    readonly_fields = (
-        "validated",
-        "status",
-        "is_locked",
-        "is_new_local_unit",
-    )
+    readonly_fields = ("status",)
     list_filter = (
+        "status",
         AutocompleteFilterFactory("Country", "country"),
         AutocompleteFilterFactory("Type", "type"),
         AutocompleteFilterFactory("Level", "level"),

local_units/bulk_upload.py

@@ -119,6 +119,10 @@ def delete_existing_local_unit(self):
     def _validate_csv(self, fieldnames) -> None:
         pass
 
+    def _is_csv_empty(self, csv_reader: csv.DictReader) -> tuple[bool, list[dict]]:
+        rows = list(csv_reader)
+        return len(rows) == 0, rows
+
     def process_row(self, data: Dict[str, Any]) -> bool:
         serializer = self.serializer_class(data=data)
         if serializer.is_valid():
@@ -133,12 +137,18 @@ def run(self) -> None:
             csv_reader = csv.DictReader(file)
             fieldnames = csv_reader.fieldnames or []
             try:
+                is_empty, rows = self._is_csv_empty(csv_reader)
+                if is_empty:
+                    raise BulkUploadError("The uploaded CSV file is empty or contains only blank rows.")
+
+                csv_reader = iter(rows)
+
                 self._validate_csv(fieldnames)
             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"])
-                logger.error(f"[BulkUpload:{self.bulk_upload.pk}] Validation error: {str(e)}")
+                logger.warning(f"[BulkUpload:{self.bulk_upload.pk}] Validation error: {str(e)}")
                 return
 
             context = self.get_context().__dict__
@@ -231,7 +241,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 BulkUploadError(f"Health data are not allowed for this type: {local_unit_type.name}.")
+            raise BulkUploadError(f"You cannot upload Healthcare data when the Local Unit type is set to {local_unit_type.name}.")
 
 
 class BulkUploadHealthData(BaseBulkUpload[LocalUnitUploadContext]):

local_units/filterset.py

@@ -22,8 +22,6 @@ class Meta:
             "country__iso",
             "type__code",
             "draft",
-            "validated",
-            "is_locked",
             "status",
         )
 

local_units/migrations/0023_remove_localunit_is_locked_and_more.py

@@ -0,0 +1,34 @@
+# Generated by Django 4.2.19 on 2025-09-02 17:20
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("local_units", "0022_localunit_status_migrate"),
+    ]
+
+    operations = [
+        migrations.RemoveField(
+            model_name="localunit",
+            name="is_locked",
+        ),
+        migrations.RemoveField(
+            model_name="localunit",
+            name="is_new_local_unit",
+        ),
+        migrations.RemoveField(
+            model_name="localunit",
+            name="validated",
+        ),
+        migrations.AlterField(
+            model_name="localunit",
+            name="status",
+            field=models.IntegerField(
+                choices=[(1, "Validated"), (2, "Unvalidated"), (3, "Pending Edit Validation"), (4, "Externally Managed")],
+                default=2,
+                verbose_name="Validation Status",
+            ),
+        ),
+    ]

local_units/models.py

@@ -355,6 +355,7 @@ class Status(models.IntegerChoices):
         VALIDATED = 1, "Validated"
         UNVALIDATED = 2, "Unvalidated"
         PENDING_EDIT_VALIDATION = 3, "Pending Edit Validation"
+        EXTERNALLY_MANAGED = 4, "Externally Managed"
 
     # added to track health local unit data (Table B)
     health = models.ForeignKey(
@@ -394,7 +395,6 @@ class Status(models.IntegerChoices):
         auto_now=False,
     )
     draft = models.BooleanField(default=False, verbose_name=_("Draft"))
-    validated = models.BooleanField(default=False, verbose_name=_("Validated"))  # NOTE: This field might be deprecated soon.
     status = models.IntegerField(
         choices=Status.choices, default=Status.UNVALIDATED, verbose_name=_("Validation Status")
     )  # NOTE: Replacement of validated field for better status tracking
@@ -422,7 +422,6 @@ class Status(models.IntegerChoices):
     email = models.EmailField(max_length=255, blank=True, null=True, verbose_name=_("Email"))
     link = models.URLField(max_length=255, blank=True, null=True, verbose_name=_("Social link"))
     location = models.PointField(srid=4326, help_text="Local Unit Location")
-    is_locked = models.BooleanField(default=False, verbose_name=_("Is locked?"))
     is_deprecated = models.BooleanField(default=False, verbose_name=_("Is deprecated?"))
     deprecated_reason = models.IntegerField(
         choices=DeprecateReason.choices, verbose_name=_("deprecated reason"), blank=True, null=True
@@ -446,8 +445,6 @@ class Status(models.IntegerChoices):
         related_name="bulk_upload_local_unit",
     )
 
-    is_new_local_unit = models.BooleanField(default=False, verbose_name=("Is New Local Unit?"))
-
     def __str__(self):
         branch_name = self.local_branch_name or self.english_branch_name
         return f"{branch_name} ({self.country.name})"

local_units/serializers.py

@@ -263,7 +263,6 @@ class PrivateLocalUnitDetailSerializer(NestedCreateMixin, NestedUpdateMixin):
     modified_by_details = LocalUnitMiniUserSerializer(source="modified_by", read_only=True)
     created_by_details = LocalUnitMiniUserSerializer(source="created_by", read_only=True)
     version_id = serializers.SerializerMethodField()
-    is_locked = serializers.BooleanField(read_only=True)
 
     class Meta:
         model = LocalUnit
@@ -306,9 +305,7 @@ class Meta:
             "modified_by_details",
             "created_by_details",
             "version_id",
-            "is_locked",
             "update_reason_overview",
-            "is_new_local_unit",
             "bulk_upload",
         )
 
@@ -374,8 +371,7 @@ def create(self, validated_data):
                 )
         validated_data["location"] = GEOSGeometry("POINT(%f %f)" % (lng, lat))
         validated_data["created_by"] = self.context["request"].user
-        validated_data["is_locked"] = True
-        validated_data["is_new_local_unit"] = True
+        validated_data["status"] = LocalUnit.Status.UNVALIDATED
         return super().create(validated_data)
 
     def update(self, instance, validated_data):
@@ -445,7 +441,6 @@ class PrivateLocalUnitSerializer(serializers.ModelSerializer):
     health_details = MiniHealthDataSerializer(read_only=True, source="health")
     status_details = serializers.CharField(source="get_status_display", read_only=True)
     modified_by_details = LocalUnitMiniUserSerializer(source="modified_by", read_only=True)
-    is_locked = serializers.BooleanField(read_only=True)
     update_reason_overview = serializers.CharField(read_only=True)
 
     class Meta:
@@ -471,8 +466,6 @@ class Meta:
             "phone",
             "modified_at",
             "modified_by_details",
-            "is_locked",
-            "is_new_local_unit",
             "bulk_upload",
             "update_reason_overview",
         )
@@ -944,9 +937,7 @@ def validate(self, validated_data):
                 )
 
         validated_data["location"] = GEOSGeometry("POINT(%f %f)" % (longitude, latitude))
-        validated_data["validated"] = True  # This might deprecate soon
-        validated_data["is_locked"] = True
-        validated_data["status"] = LocalUnit.Status.VALIDATED
+        validated_data["status"] = LocalUnit.Status.EXTERNALLY_MANAGED
 
         # NOTE: Bulk upload doesn't call create() method
         health_data = validated_data.pop("health", None)