Merge pull request #2572 from IFRCGo/feature/api-endpoint-for-health-units

Endpoint and serializer for Health units

Author: szabozoltan69

local_units/filterset.py

@@ -57,3 +57,29 @@ class Meta:
             "country__iso": ["exact", "in"],
             "country__id": ["exact", "in"],
         }
+
+
+class HealthLocalUnitFilters(filters.FilterSet):
+    # Simple filters for health-local-units endpoint
+    region = filters.NumberFilter(field_name="country__region_id", label="Region")
+    country = filters.NumberFilter(field_name="country_id", label="Country")
+    iso3 = filters.CharFilter(field_name="country__iso3", lookup_expr="exact", label="ISO3")
+    validated = filters.BooleanFilter(method="filter_validated", label="Validated")
+    subtype = filters.CharFilter(field_name="subtype", lookup_expr="icontains", label="Subtype")
+
+    class Meta:
+        model = LocalUnit
+        fields = (
+            "region",
+            "country",
+            "iso3",
+            "validated",
+            "subtype",
+        )
+
+    def filter_validated(self, queryset, name, value):
+        if value is True:
+            return queryset.filter(status=LocalUnit.Status.VALIDATED)
+        if value is False:
+            return queryset.exclude(status=LocalUnit.Status.VALIDATED)
+        return queryset

local_units/serializers.py

@@ -945,3 +945,176 @@ def validate(self, validated_data):
             health_instance = HealthData.objects.create(**health_data)
             validated_data["health"] = health_instance
         return validated_data
+
+
+# Public, flattened serializer for Health Local Units (Type Code = 2)
+class _CodeNameSerializer(serializers.Serializer):
+    code = serializers.IntegerField()
+    name = serializers.CharField()
+
+
+class HealthLocalUnitFlatSerializer(serializers.ModelSerializer):
+    # LocalUnit basics
+    id = serializers.IntegerField(read_only=True)
+    country_id = serializers.IntegerField(source="country.id", read_only=True)
+    country_name = serializers.CharField(source="country.name", read_only=True)
+    country_iso3 = serializers.CharField(source="country.iso3", read_only=True)
+    type_code = serializers.IntegerField(source="type.code", read_only=True)
+    type_name = serializers.CharField(source="type.name", read_only=True)
+    status_display = serializers.CharField(source="get_status_display", read_only=True)
+    location = serializers.SerializerMethodField()
+
+    # HealthData flattened
+    affiliation = serializers.SerializerMethodField()
+    functionality = serializers.SerializerMethodField()
+    health_facility_type = serializers.SerializerMethodField()
+    primary_health_care_center = serializers.SerializerMethodField()
+    hospital_type = serializers.SerializerMethodField()
+
+    general_medical_services = serializers.SerializerMethodField()
+    specialized_medical_beyond_primary_level = serializers.SerializerMethodField()
+    blood_services = serializers.SerializerMethodField()
+    professional_training_facilities = serializers.SerializerMethodField()
+
+    class Meta:
+        model = LocalUnit
+        fields = (
+            # LocalUnit
+            "id",
+            "country_id",
+            "country_name",
+            "country_iso3",
+            "local_branch_name",
+            "english_branch_name",
+            "address_loc",
+            "address_en",
+            "city_loc",
+            "city_en",
+            "postcode",
+            "phone",
+            "email",
+            "link",
+            "focal_person_loc",
+            "focal_person_en",
+            "date_of_data",
+            "subtype",
+            "type_code",
+            "type_name",
+            "status",
+            "status_display",
+            "location",
+            # HealthData core
+            "affiliation",
+            "other_affiliation",
+            "functionality",
+            "focal_point_email",
+            "focal_point_phone_number",
+            "focal_point_position",
+            "health_facility_type",
+            "other_facility_type",
+            "primary_health_care_center",
+            "speciality",
+            "hospital_type",
+            "is_teaching_hospital",
+            "is_in_patient_capacity",
+            "is_isolation_rooms_wards",
+            "maximum_capacity",
+            "number_of_isolation_rooms",
+            "is_warehousing",
+            "is_cold_chain",
+            "ambulance_type_a",
+            "ambulance_type_b",
+            "ambulance_type_c",
+            "general_medical_services",
+            "specialized_medical_beyond_primary_level",
+            "other_services",
+            "blood_services",
+            "professional_training_facilities",
+            "total_number_of_human_resource",
+            "general_practitioner",
+            "specialist",
+            "residents_doctor",
+            "nurse",
+            "dentist",
+            "nursing_aid",
+            "midwife",
+            "other_medical_heal",
+            "other_profiles",
+            "feedback",
+        )
+
+    # NOTE: HealthData direct field mappings via source
+    other_affiliation = serializers.CharField(source="health.other_affiliation", read_only=True)
+    focal_point_email = serializers.EmailField(source="health.focal_point_email", read_only=True)
+    focal_point_phone_number = serializers.CharField(source="health.focal_point_phone_number", read_only=True)
+    focal_point_position = serializers.CharField(source="health.focal_point_position", read_only=True)
+    other_facility_type = serializers.CharField(source="health.other_facility_type", read_only=True)
+    speciality = serializers.CharField(source="health.speciality", read_only=True)
+    is_teaching_hospital = serializers.BooleanField(source="health.is_teaching_hospital", read_only=True)
+    is_in_patient_capacity = serializers.BooleanField(source="health.is_in_patient_capacity", read_only=True)
+    is_isolation_rooms_wards = serializers.BooleanField(source="health.is_isolation_rooms_wards", read_only=True)
+    maximum_capacity = serializers.IntegerField(source="health.maximum_capacity", read_only=True)
+    number_of_isolation_rooms = serializers.IntegerField(source="health.number_of_isolation_rooms", read_only=True)
+    is_warehousing = serializers.BooleanField(source="health.is_warehousing", read_only=True)
+    is_cold_chain = serializers.BooleanField(source="health.is_cold_chain", read_only=True)
+    ambulance_type_a = serializers.IntegerField(source="health.ambulance_type_a", read_only=True)
+    ambulance_type_b = serializers.IntegerField(source="health.ambulance_type_b", read_only=True)
+    ambulance_type_c = serializers.IntegerField(source="health.ambulance_type_c", read_only=True)
+    other_services = serializers.CharField(source="health.other_services", read_only=True)
+    total_number_of_human_resource = serializers.IntegerField(source="health.total_number_of_human_resource", read_only=True)
+    general_practitioner = serializers.IntegerField(source="health.general_practitioner", read_only=True)
+    specialist = serializers.IntegerField(source="health.specialist", read_only=True)
+    residents_doctor = serializers.IntegerField(source="health.residents_doctor", read_only=True)
+    nurse = serializers.IntegerField(source="health.nurse", read_only=True)
+    dentist = serializers.IntegerField(source="health.dentist", read_only=True)
+    nursing_aid = serializers.IntegerField(source="health.nursing_aid", read_only=True)
+    midwife = serializers.IntegerField(source="health.midwife", read_only=True)
+    other_medical_heal = serializers.BooleanField(source="health.other_medical_heal", read_only=True)
+    other_profiles = serializers.CharField(source="health.other_profiles", read_only=True)
+    feedback = serializers.CharField(source="health.feedback", read_only=True)
+
+    def get_location(self, unit) -> dict:
+        return {"lat": unit.location.y, "lng": unit.location.x}
+
+    def _code_name(self, obj):
+        if not obj:
+            return None
+        return {"code": obj.code, "name": obj.name}
+
+    def _code_name_list(self, qs):
+        return [{"code": x.code, "name": x.name} for x in qs.all()] if qs is not None else []
+
+    def get_affiliation(self, obj):
+        return self._code_name(getattr(obj.health, "affiliation", None) if obj.health else None)
+
+    def get_functionality(self, obj):
+        return self._code_name(getattr(obj.health, "functionality", None) if obj.health else None)
+
+    def get_health_facility_type(self, obj):
+        if not obj.health or not obj.health.health_facility_type:
+            return None
+        ft = obj.health.health_facility_type
+        data = {"code": ft.code, "name": ft.name}
+        # Attach image_url if request in context
+        request = self.context.get("request") if self.context else None
+        if request:
+            data["image_url"] = FacilityType.get_image_map(ft.code, request)
+        return data
+
+    def get_primary_health_care_center(self, obj):
+        return self._code_name(getattr(obj.health, "primary_health_care_center", None) if obj.health else None)
+
+    def get_hospital_type(self, obj):
+        return self._code_name(getattr(obj.health, "hospital_type", None) if obj.health else None)
+
+    def get_general_medical_services(self, obj):
+        return self._code_name_list(obj.health.general_medical_services if obj.health else None)
+
+    def get_specialized_medical_beyond_primary_level(self, obj):
+        return self._code_name_list(obj.health.specialized_medical_beyond_primary_level if obj.health else None)
+
+    def get_blood_services(self, obj):
+        return self._code_name_list(obj.health.blood_services if obj.health else None)
+
+    def get_professional_training_facilities(self, obj):
+        return self._code_name_list(obj.health.professional_training_facilities if obj.health else None)

local_units/test_views.py

@@ -1580,3 +1580,157 @@ def test_empty_health_template_file(cls):
         cls.assertIsNotNone(cls.bulk_upload.error_message)
         cls.assertEqual(LocalUnit.objects.count(), 5)
         cls.assertEqual(HealthData.objects.count(), 5)
+
+
+class TestHealthLocalUnitsPublicList(APITestCase):
+    """
+    Tests for the public, flattened health local units endpoint: /api/v2/health-local-units/
+    Only add new tests; existing code remains untouched.
+    """
+
+    def setUp(self):
+        super().setUp()
+        # Regions and countries
+        self.region1 = RegionFactory.create(name=2, label="Asia Pacific")
+        self.region2 = RegionFactory.create(name=1, label="Americas")
+
+        self.country1 = CountryFactory.create(name="Nepal", iso3="NPL", region=self.region1)
+        self.country2 = CountryFactory.create(name="Philippines", iso3="PHL", region=self.region1)
+        self.country3 = CountryFactory.create(name="Brazil", iso3="BRA", region=self.region2)
+
+        # Types
+        self.type_health = LocalUnitType.objects.create(code=2, name="Health")
+        self.type_admin = LocalUnitType.objects.create(code=1, name="Administrative")
+
+        # Lookups for HealthData
+        self.aff = Affiliation.objects.create(code=11, name="Public")
+        self.func = Functionality.objects.create(code=21, name="Functional")
+        self.ftype = FacilityType.objects.create(code=31, name="Clinic")
+        self.phcc = PrimaryHCC.objects.create(code=41, name="Primary")
+        self.htype = HospitalType.objects.create(code=51, name="District Hospital")
+
+        # Included: public, not deprecated, type=2 with health
+        self.hd1 = HealthDataFactory.create(
+            affiliation=self.aff,
+            functionality=self.func,
+            health_facility_type=self.ftype,
+            primary_health_care_center=self.phcc,
+            hospital_type=self.htype,
+        )
+        self.lu1 = LocalUnitFactory.create(
+            country=self.country1,
+            type=self.type_health,
+            health=self.hd1,
+            visibility=VisibilityChoices.PUBLIC,
+            is_deprecated=False,
+            status=LocalUnit.Status.VALIDATED,
+            subtype="District Clinic A",
+        )
+
+        self.hd2 = HealthDataFactory.create(
+            affiliation=self.aff,
+            functionality=self.func,
+            health_facility_type=self.ftype,
+        )
+        self.lu2 = LocalUnitFactory.create(
+            country=self.country2,
+            type=self.type_health,
+            health=self.hd2,
+            visibility=VisibilityChoices.PUBLIC,
+            is_deprecated=False,
+            status=LocalUnit.Status.UNVALIDATED,
+            subtype="Mobile Clinic",
+        )
+
+        # Exclusions
+        # - private visibility
+        LocalUnitFactory.create(
+            country=self.country1,
+            type=self.type_health,
+            health=HealthDataFactory.create(affiliation=self.aff, functionality=self.func, health_facility_type=self.ftype),
+            visibility=VisibilityChoices.MEMBERSHIP,
+            is_deprecated=False,
+            status=LocalUnit.Status.VALIDATED,
+        )
+        # - deprecated
+        LocalUnitFactory.create(
+            country=self.country1,
+            type=self.type_health,
+            health=HealthDataFactory.create(affiliation=self.aff, functionality=self.func, health_facility_type=self.ftype),
+            visibility=VisibilityChoices.PUBLIC,
+            is_deprecated=True,
+            status=LocalUnit.Status.VALIDATED,
+        )
+        # - wrong type (admin)
+        LocalUnitFactory.create(
+            country=self.country1,
+            type=self.type_admin,
+            health=None,
+            visibility=VisibilityChoices.PUBLIC,
+            is_deprecated=False,
+            status=LocalUnit.Status.VALIDATED,
+        )
+        # - no health
+        LocalUnitFactory.create(
+            country=self.country3,
+            type=self.type_health,
+            health=None,
+            visibility=VisibilityChoices.PUBLIC,
+            is_deprecated=False,
+            status=LocalUnit.Status.VALIDATED,
+        )
+
+    def test_list_public_health_local_units(self):
+        resp = self.client.get("/api/v2/health-local-units/")
+        self.assertEqual(resp.status_code, 200)
+        self.assertEqual(resp.data["count"], 2)
+        # Check a few flattened fields exist
+        first = resp.data["results"][0]
+        self.assertIn("country_name", first)
+        self.assertIn("country_iso3", first)
+        self.assertEqual(first["type_code"], 2)
+        self.assertIn("location", first)
+        self.assertIn("affiliation", first)
+        self.assertIn("functionality", first)
+        self.assertIn("health_facility_type", first)
+        # health_facility_type may include image_url; assert at least name exists when present
+        if first["health_facility_type"] is not None:
+            self.assertIn("name", first["health_facility_type"])
+
+    def test_filters_region_country_iso3_validated_subtype(self):
+        # region -> both country1 and country2 are in region1
+        resp = self.client.get(f"/api/v2/health-local-units/?region={self.region1.id}")
+        self.assertEqual(resp.status_code, 200)
+        self.assertEqual(resp.data["count"], 2)
+
+        # country
+        resp = self.client.get(f"/api/v2/health-local-units/?country={self.country1.id}")
+        self.assertEqual(resp.status_code, 200)
+        self.assertEqual(resp.data["count"], 1)
+        self.assertEqual(resp.data["results"][0]["country_iso3"], "NPL")
+
+        # iso3
+        resp = self.client.get("/api/v2/health-local-units/?iso3=PHL")
+        self.assertEqual(resp.status_code, 200)
+        self.assertEqual(resp.data["count"], 1)
+        self.assertEqual(resp.data["results"][0]["country_iso3"], "PHL")
+
+        # validated true
+        resp = self.client.get("/api/v2/health-local-units/?validated=true")
+        self.assertEqual(resp.status_code, 200)
+        self.assertEqual(resp.data["count"], 1)
+        self.assertEqual(resp.data["results"][0]["status"], LocalUnit.Status.VALIDATED)
+
+        # validated false
+        resp = self.client.get("/api/v2/health-local-units/?validated=false")
+        self.assertEqual(resp.status_code, 200)
+        self.assertEqual(resp.data["count"], 1)
+        self.assertEqual(resp.data["results"][0]["status"], LocalUnit.Status.UNVALIDATED)
+
+        # subtype icontains
+        resp = self.client.get("/api/v2/health-local-units/?subtype=mobile")
+        self.assertEqual(resp.status_code, 200)
+        self.assertEqual(resp.data["count"], 1)
+        self.assertEqual(resp.data["results"][0]["subtype"].lower(), "mobile clinic".lower())
+
+        # End of relevant assertions for this test.