Feature/filter administrative divisions by address

requirements.in

@@ -43,4 +43,5 @@ pyshp
 polyline
 drf-spectacular
 xmltodict
-freezegun
+freezegun
+geopy

requirements.txt

@@ -110,6 +110,10 @@ flake8-polyfill==1.0.2
     # via pep8-naming
 freezegun==1.5.1
     # via -r requirements.in
+geographiclib==2.0
+    # via geopy
+geopy==2.4.1
+    # via -r requirements.in
 idna==3.7
     # via requests
 inflection==0.5.1

services/api.py

@@ -42,6 +42,7 @@
 )
 from services.models.unit import CONTRACT_TYPES, ORGANIZER_TYPES, PROVIDER_TYPES
 from services.utils import check_valid_concrete_field
+from services.utils.geocode_address import geocode_address
 
 if settings.REST_FRAMEWORK and settings.REST_FRAMEWORK["DEFAULT_RENDERER_CLASSES"]:
     DEFAULT_RENDERERS = [
@@ -1162,6 +1163,24 @@ def to_representation(self, obj):
 class AdministrativeDivisionViewSet(munigeo_api.AdministrativeDivisionViewSet):
     serializer_class = AdministrativeDivisionSerializer
 
+    def get_queryset(self):
+        queryset = super().get_queryset()
+        filters = self.request.query_params
+
+        if "address" in filters and "municipality" in filters:
+            street_address = filters["address"]
+            municipality = filters["municipality"]
+            country = settings.DEFAULT_COUNTRY
+            address = f"{street_address}, {municipality}, {country}"
+            location_coordinates = geocode_address(address)
+            if location_coordinates:
+                point = Point(
+                    location_coordinates[1], location_coordinates[0], srid=4326
+                )
+                queryset = queryset.filter(geometry__boundary__contains=point)
+
+        return queryset.order_by("id")
+
 
 register_view(AdministrativeDivisionViewSet, "administrative_division")
 

services/tests/test_administrative_division_view_set_api.py

@@ -0,0 +1,106 @@
+import pytest
+from django.conf import settings
+from django.contrib.gis.geos import MultiPolygon, Polygon
+from django.urls import reverse
+from munigeo.models import (
+    AdministrativeDivision,
+    AdministrativeDivisionGeometry,
+    AdministrativeDivisionType,
+    Municipality,
+)
+from rest_framework.test import APIClient
+
+from services.api import make_muni_ocd_id
+from services.tests.utils import get
+
+
+def create_administrative_divisions():
+    municipality_ids = ["helsinki", "espoo", "vantaa"]
+    division_type = AdministrativeDivisionType.objects.create(type="muni")
+    for municipality_id in municipality_ids:
+        municipality = Municipality.objects.create(
+            id=municipality_id, name=municipality_id
+        )
+        AdministrativeDivision.objects.create(
+            type=division_type,
+            name=municipality_id,
+            ocd_id=make_muni_ocd_id(municipality_id),
+            municipality=municipality,
+        )
+
+
+def create_test_area():
+    """
+    Create a simple test area in Helsinki center.
+    """
+    polygon_coords = [
+        (24.928, 60.178),  # top left
+        (24.948, 60.178),  # top right
+        (24.948, 60.159),  # bottom right
+        (24.928, 60.159),  # bottom left
+        (24.928, 60.178),  # Close the ring by repeating the first point
+    ]
+    polygon = Polygon(polygon_coords, srid=4326)  # WGS84 srid
+    multi_polygon = MultiPolygon(polygon, srid=4326)
+    multi_polygon.transform(settings.DEFAULT_SRID)
+    return multi_polygon
+
+
+@pytest.fixture
+def api_client():
+    return APIClient()
+
+
+@pytest.mark.django_db
+def test_get_administrative_division_list(api_client):
+    create_administrative_divisions()
+    response = get(api_client, reverse("administrativedivision-list"))
+    assert response.status_code == 200
+    assert response.data["count"] == 3
+
+
+@pytest.mark.django_db
+def test_municipality_filter(api_client):
+    create_administrative_divisions()
+    response = get(
+        api_client,
+        reverse("administrativedivision-list"),
+        data={"municipality": "helsinki"},
+    )
+    assert response.status_code == 200
+    assert response.data["count"] == 1
+    assert response.data["results"][0]["municipality"] == "helsinki"
+
+
+@pytest.mark.django_db
+def test_address_filter(api_client):
+    create_administrative_divisions()
+    division = AdministrativeDivision.objects.get(name="helsinki")
+    AdministrativeDivisionGeometry.objects.create(
+        division=division, boundary=create_test_area()
+    )
+
+    response = get(
+        api_client,
+        reverse("administrativedivision-list"),
+        data={
+            "municipality": "helsinki",
+            "address": "Kaivokatu 1",
+        },  # An address in the test area
+    )
+
+    assert response.status_code == 200
+    assert response.data["count"] == 1
+    assert response.data["results"][0]["municipality"] == "helsinki"
+
+    response = get(
+        api_client,
+        reverse("administrativedivision-list"),
+        data={
+            "municipality": "helsinki",
+            "address": "Katajanokanranta 1",
+        },  # An address outside the test area
+    )
+
+    assert response.status_code == 200
+    assert response.data["count"] == 0

services/utils/geocode_address.py

@@ -0,0 +1,12 @@
+from geopy.geocoders import Nominatim
+
+
+def geocode_address(address):
+    """
+    Geocodes address and returns location coordinates.
+    """
+    geolocator = Nominatim(user_agent="smbackend")
+    location = geolocator.geocode(address)
+    if location:
+        return location.latitude, location.longitude
+    return None